Nacos源码-从Demo出发研究事件驱动与观察者模式的应用

news2025/1/12 13:33:17

在我们分析 Nacos 源码时，会看见大量的事件发布的动作，不管是客户端注册/下线、服务改变、服务订阅等等都是利用了事件发布。

下面我在自己的项目中，引入Nacos的依赖进行一个简单的demo的演示，我个人认为其和spring容器的listener也有想通之处：

导入的依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.jxz</groupId>
    <artifactId>NacosListenerTest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.12.RELEASE</version>
    </parent>

    <properties>
        <java.version>1.8</java.version>
        <spring-cloud.version>Hoxton.SR12</spring-cloud.version>
        <spring-cloud-alibaba.version>2.2.8.RELEASE</spring-cloud-alibaba.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <!--整合spring cloud-->
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <!--整合spring cloud alibaba-->
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>${spring-cloud-alibaba.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>


    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>


</project>

1、首先我们需要自己定义一个事件，直接继承com.alibaba.nacos.common.notify.Event：

package com.jxz.nacos;

import com.alibaba.nacos.common.notify.Event;

/**
 * @Author jiangxuzhao
 * @Description 自定义的事件
 * @Date 2023/5/28
 */
public class JXZNacosEvent extends Event {

}
  1. 实现一个订阅者来处理发布的事件，需要继承com.alibaba.nacos.common.notify.listener.SmartSubscriber;：
package com.jxz.nacos;

import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author jiangxuzhao
 * @Description 事件订阅者
 * @Date 2023/5/28
 */
@Component
public class JXZNacosSubscriber extends SmartSubscriber {
    public JXZNacosSubscriber() {
        NotifyCenter.registerSubscriber(this);
    }

    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        List<Class<? extends Event>> list = new ArrayList<>();
        list.add(JXZNacosEvent.class);
        return list;
    }

    @Override
    public void onEvent(Event event) {
        System.out.println("订阅者收到了事件的类型为："+event.getClass());
    }
}

这里有三个方法：

a) 构造方法中，需要注册订阅者到Nacos的通知中心去

b) 重写subscribeTypes()方法，把需要订阅的事件放进去

c) 重写onEvent()方法， 主要是订阅者接到事件后的处理动作

  1. 起一个SpringBoot项目测试一下，在里面发布一个JXZNacosEvent事件：
package com.jxz;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.jxz.nacos.JXZNacosEvent;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

/**
 * @Author jiangxuzhao
 * @Description
 * @Date 2023/5/28
 */
@SpringBootApplication
public class NacosListenerApp {
    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(NacosListenerApp.class, args);
        NotifyCenter.publishEvent(new JXZNacosEvent());
    }
}

打印结果如下，显然是按照onEvent(Event event)方法中的逻辑开展了：
在这里插入图片描述

前面有一些报错信息，主要是和Nacos服务的配置有关，我这里为了demo没写。

接下来看看他的源码。

事件发布源码分析

从NotifyCenter.publishEvent(new JXZNacosEvent());开始分析：

NotifyCenter.publishEvent(new JXZNacosEvent());


public static boolean publishEvent(final Event event) {
    try {
        // 主线任务
        return publishEvent(event.getClass(), event);
    } catch (Throwable ex) {
        LOGGER.error("There was an exception to the message publishing : ", ex);
        return false;
    }
}

private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
    if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
        return INSTANCE.sharePublisher.publish(event);
    }
    
    final String topic = ClassUtils.getCanonicalName(eventType);
    // 根据发布事件类型获取 EventPublisher 对象，该对象中会包含订阅者信息
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher != null) {
        // 跟主线任务
        return publisher.publish(event);
    }
    LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    return false;
}

@Override
public boolean publish(Event event) {
    checkIsStart();
    // 把事件放入到了一个 阻塞队列
    boolean success = this.queue.offer(event);
    if (!success) {
        LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
        receiveEvent(event);
        return true;
    }
    return true;
}

看到阻塞队列，肯定后台有一个异步任务，在从阻塞队列取数据，相当于是发布事件的任务被异步化了：

在这里插入图片描述

定位到com.alibaba.nacos.common.notify.DefaultPublisher#openEventHandler中的下面的代码：

for (; ; ) {
    if (shutdown) {
        break;
    }
    // 从阻塞队列取数据
    final Event event = queue.take();
    // 处理事件，通知相关的订阅者
    receiveEvent(event);
    UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));
}

这种 阻塞队列+异步任务的方式在Nacos1.4.1中也有用到，同时往上看到openEventHandler方法写在重写的com.alibaba.nacos.common.notify.DefaultPublisher#run方法中：

@Override
public void run() {
    openEventHandler();
}

一下子联想到线程创建的方式，果然是通过extend Thread的方式定义的线程类：

public class DefaultPublisher extends Thread implements EventPublisher {}

继续回到刚刚的阻塞队列取出处理事件那里：

protected final ConcurrentHashSet<Subscriber> subscribers = new ConcurrentHashSet<>();

void receiveEvent(Event event) {
    final long currentEventSequence = event.sequence();
    
    if (!hasSubscriber()) {
        LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);
        return;
    }
    
    // 循环遍历通知发布事件对应的订阅者
    for (Subscriber subscriber : subscribers) {
        // Whether to ignore expiration events
        if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {
            LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",
                    event.getClass());
            continue;
        }
        
        // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.
        // Remove original judge part of codes.
        //回调通知
        notifySubscriber(subscriber, event);
    }
}

这里其实就是观察者模式的一大应用，参考我之前写的一篇帮助理解的文章->简单Demo理解观察者模式

subscribers就是订阅了当前事件的订阅者，不会通知全部事件的订阅者。

然后最后调用了 notifySubscriber(subscriber, event); 方法来通知订阅者，并让订阅者执行，代码如下:

@Override
public void notifySubscriber(final Subscriber subscriber, final Event event) {
    
    LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
    
    // 创建一个Runnable任务，就是调用订阅者的 onEvent 方法
    final Runnable job = () -> subscriber.onEvent(event);
    // 拿到执行器
    final Executor executor = subscriber.executor();
    
    if (executor != null) {
        // 执行任务
        executor.execute(job);
    } else {
        try {
            job.run();
        } catch (Throwable e) {
            LOGGER.error("Event callback exception: ", e);
        }
    }
}

通过上面的分析我们可以得到，事件发布是通过 阻塞队列+异步任务的方式来实现的，订阅者执行onEvent()里的任务。

订阅者注册源码分析

订阅者注册从com.jxz.nacos.JXZNacosSubscriber#JXZNacosSubscriber开始分析，代码如下：

public TestSubscriber() {
    NotifyCenter.registerSubscriber(this);
}

public static void registerSubscriber(final Subscriber consumer) {
    registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
}

public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
    // If you want to listen to multiple events, you do it separately,
    // based on subclass's subscribeTypes method return list, it can register to publisher.
    if (consumer instanceof SmartSubscriber) {
        // 在这里会调用 subscribeTypes 方法，来获取我们需要监听的事件，然后进行遍历
        for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
            // For case, producer: defaultSharePublisher -> consumer: smartSubscriber.
            if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
            } else {
                // For case, producer: defaultPublisher -> consumer: subscriber.
                // 添加订阅者，主线任务
                addSubscriber(consumer, subscribeType, factory);
            }
        }
        return;
    }
    
    // 这里代码省略
}

上面就用到了com.jxz.nacos.JXZNacosSubscriber#subscribeTypes方法，对这些事件进行遍历监听，主要是获取需要监听那些事件，然后调用 addSubscriber() 方法，将事件与订阅者进行关系绑定，代码如下：

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,
        EventPublisherFactory factory) {
    
    final String topic = ClassUtils.getCanonicalName(subscribeType);
    synchronized (NotifyCenter.class) {
        // 这里很关键，会创建 EventPublisher 对象，一个Event事件会对应一个 EventPublisher 一个对象
        MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);
    }
    // 获取事件对应的 EventPublisher 对象，我这里是com.jxz.nacos.JXZNacosEvent
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher instanceof ShardedEventPublisher) {
        ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);
    } else {
        // 往事件对应的 EventPublisher 对象中添加订阅者信息
        publisher.addSubscriber(consumer);
    }
}

通过debug看以下内容：
在这里插入图片描述

MapUtil.computeIfAbsent完成的动作是，对每一个Event事件，创建一个对应的EventPublisher对象，这个对象中包含着订阅者信息，也就是一个ConcurrentHashSet<Subscriber> subscribers列表。

com.alibaba.nacos.common.utils.MapUtil#computeIfAbsent源码如下：

@NotThreadSafe
    public static <K, C, V, T> V computeIfAbsent(Map<K, V> target, K key, BiFunction<C, T, V> mappingFunction, C param1,
            T param2) {
        
        Objects.requireNonNull(target, "target");
        Objects.requireNonNull(key, "key");
        Objects.requireNonNull(mappingFunction, "mappingFunction");
        Objects.requireNonNull(param1, "param1");
        Objects.requireNonNull(param2, "param2");
        
        V val = target.get(key);
        if (val == null) {
            V ret = mappingFunction.apply(param1, param2);
            target.put(key, ret);
            return ret;
        }
        return val;
    }

首先有一个INSTANCE.publisherMap集合，整体结构是这样的，都是内部的成员变量，饿汉式地在内部指定了一个INSTANCE变量，其包含一个publisherMap容器：

public class NotifyCenter {
  private static final NotifyCenter INSTANCE = new NotifyCenter();
   /**
     * Publisher management container.
     */
  private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<>(16);
}

然后就是用一个com.alibaba.nacos.common.notify.NotifyCenter#DEFAULT_PUBLISHER_FACTORY中默认的工厂方法，去根据传入的subscribeType创建对应的对象作为publisherMap容器的value，key为topic.

DEFAULT_PUBLISHER_FACTORY= (cls, buffer) -> {
    try {
        EventPublisher publisher = clazz.newInstance();
        publisher.init(cls, buffer);
        return publisher;
    } catch (Throwable ex) {
        LOGGER.error("Service class newInstance has error : ", ex);
        throw new NacosRuntimeException(*SERVER_ERROR*, ex);
    }
};

最后 publisher.addSubscriber(consumer);其实调用的是com.alibaba.nacos.common.notify.DefaultPublisher#addSubscriber，往DefaultPublisher中的成员变量subscribers加入新的监听者，这表示的是一个事件对应的那些在监听它的Subscriber。

protected final ConcurrentHashSet<Subscriber> subscribers = new ConcurrentHashSet<>();

这也是之前com.alibaba.nacos.common.notify.DefaultPublisher#receiveEvent进行事件通知的时候，需要进行遍历，然后回调通知的那个变量。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/580078.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Koa学习1：初始化项目

Koa学习1：初始化项目

前言 作为前端开发者&#xff0c;最适合我们的后端就是node了&#xff0c;node的框架挺多的。选择Koa是因为国内用的挺多的、关于这方面的教程也很多、而且比较适合小项目。 学习教程是&#xff1a;【杰哥课堂】-项目实战-NodeKoa2从零搭建通用API服务 写这些文章&#xff0…
阅读更多...
K8s in Action 阅读笔记——【5】Services: enabling clients to discover and talk to pods

K8s in Action 阅读笔记——【5】Services: enabling clients to discover and talk to pods

K8s in Action 阅读笔记——【5】Services: enabling clients to discover and talk to pods 你已了解Pod以及如何通过ReplicaSets等资源部署它们以确保持续运行。虽然某些Pod可以独立完成工作&#xff0c;但现今许多应用程序需要响应外部请求。例如&#xff0c;在微服务的情况…
阅读更多...
在Python中载入大量图片型数据集，与matlab结合使用时，如何解决RAM的占用爆炸性增长的问题

在Python中载入大量图片型数据集，与matlab结合使用时，如何解决RAM的占用爆炸性增长的问题

在Python中载入大量图片时&#xff0c;由于每张图片都会被转换成Numpy数组并存储在内存中&#xff0c;因此可能会导致RAM的占用爆炸性增长。为了减少RAM的使用&#xff0c;可以考虑采用以下方法&#xff1a; Python和Matlab结合使用。首先&#xff0c;可以使用Python的Pillow库…
阅读更多...
【Linux】遇事不决，可先点灯，LED驱动的进化之路---1

【Linux】遇事不决，可先点灯，LED驱动的进化之路---1

【Linux】遇事不决&#xff0c;可先点灯&#xff0c;LED驱动的进化之路---1 前言&#xff1a; 一、最简单的LED驱动程序 1.1 字符设备驱动程序框架 1.2 程序实战 1.2.1 驱动程序&#xff08;led_drive_simple.c&#xff09; 1.2.2 应用程序&#xff08;led_test_simple.c…
阅读更多...
C#，码海拾贝（25）——求解“三对角线方程组”的“追赶法”之C#源代码，《C#数值计算算法编程》源代码升级改进版

C#，码海拾贝（25）——求解“三对角线方程组”的“追赶法”之C#源代码，《C#数值计算算法编程》源代码升级改进版

using System; namespace Zhou.CSharp.Algorithm { /// <summary> /// 求解线性方程组的类 LEquations /// 原作 周长发 /// 改编 深度混淆 /// </summary> public static partial class LEquations { /// <summary> /…
阅读更多...
Apache Kafka - 理解Kafka内部原理

Apache Kafka - 理解Kafka内部原理

文章目录 Kafka的实现机制1. 集群成员关系&#xff1a;2. 控制器*&#xff1a;3. Kafka的复制&#xff1a;4. 请求处理&#xff1a;5. 物理存储&#xff1a; 导图 Kafka的实现机制 作为Kafka专家&#xff0c;我很高兴为您深入解释Kafka的实现机制。我将从以下几个方面对Kafka进…
阅读更多...
ARM体系结构与异常处理

ARM体系结构与异常处理

目录 一、ARM体系架构 1、ARM公司概述 ARM的含义 ARM公司 2.ARM产品系列 3.指令、指令集 指令 指令集 ARM指令集 ARM指令集 Thumb指令集 &#xff08;属于ARM指令集&#xff09; 4.编译原理 5.ARM数据类型 字节序 大端对齐 小端对齐 …
阅读更多...
VTK安装和运行

VTK安装和运行

创建日期: 2019-04-02 09:19:00 开始 学习资源 官方网站&#xff1a;https://vtk.org/ GitHub&#xff1a;https://github.com/Kitware/VTK 官方教程&#xff1a;https://vtk.org/Wiki/VTK/Tutorials 官方文档&#xff1a;https://vtk.org/documentation/ 用户手册&#…
阅读更多...
RocketMQ 学习教程——（一）安装 RocketMQ

RocketMQ 学习教程——（一）安装 RocketMQ

文章目录 RocketMQ 安装下载安装上传服务器配置环境变量修改 runserver.sh修改 runbroker.sh修改 broker.conf启动 安装 RocketMQ 控制台安装Linux 防火墙命令 Docker 安装 RocketMQ拉取镜像启动 NameServer 服务启动 Broker 服务启动控制台 RocketMQ 官网&#xff1a; http://…
阅读更多...
​【编写UI自动化测试集】Appium+Python+Unittest+HTMLRunner​

​【编写UI自动化测试集】Appium+Python+Unittest+HTMLRunner​

简介 获取AppPackage和AppActivity 定位UI控件的工具 脚本结构 PageObject分层管理 HTMLTestRunner生成测试报告 启动appium server服务 以python文件模式执行脚本生成测试报告 下载与安装 下载需要自动化测试的App并安装到手机 获取AppPackage和AppActivity 方法一 有源码的…
阅读更多...
算法11.从暴力递归到动态规划4

算法11.从暴力递归到动态规划4

算法|11.从暴力递归到动态规划4 1.最长公共子序列 题意&#xff1a;给定两个字符串str1和str2&#xff0c;返回这两个字符串的最长公共子序列长度 比如 &#xff1a; str1 “a12b3c456d”,str2 “1ef23ghi4j56k” 最长公共子序列是“123456”&#xff0c;所以返回长度6 解…
阅读更多...
【PowerShell】PowerShell 7.1 之后版本的安装

【PowerShell】PowerShell 7.1 之后版本的安装

当前以下操作系统支持PowerShell 7.1 版本的安装,非Windows 系统支持的版本和要求有一定的限制。 Windows 8.1/10 (including ARM64)Windows Server 2012 R2, 2016, 2019, and Semi-Annual Channel (SAC)Ubuntu 16.04/18.04/20.04 (including ARM64)Ubuntu 19.10 (via Snap pa…
阅读更多...
图的邻接矩阵表示

图的邻接矩阵表示

设图有n个顶点&#xff0c;则邻接矩阵是一个n*n的方阵&#xff1b;若2个顶点之间有边&#xff0c;则方阵对应位置的值为1&#xff0c;否则为0&#xff1b; 看几个例子&#xff1b; 此图的邻接矩阵是 0 1 1 1 1 0 1 0 1 1 0 1 1 0…
阅读更多...
学习 xss+csrf 组合拳

学习 xss+csrf 组合拳

目录 1.xss基础铺垫 1.1反射型xss 1.2存储型xss 1.3基于DOM的xss 1.4xss漏洞的危害 1.5xss漏洞的黑盒测试 1.6xss漏洞的白盒测试 2.csrf基础铺垫 2.1csrf攻击原理 2.2csrf攻击防护 3.应用案例 3.1存储型xsscsrf组合拳 3.2csrfselfxss组合拳 1.xss基础铺垫 跨站脚…
阅读更多...
线程和进程

线程和进程

进程和线程的区别(超详细) 与进程不同的是同类的多个线程共享进程的堆和方法区资源&#xff0c;但每个线程有自己的程序计数器、虚拟机栈和本地方法栈&#xff0c;所以系统在产生一个线程&#xff0c;或是在各个线程之间作切换工作时&#xff0c;负担要比进程小得多&#xff0…
阅读更多...
【架构】常见技术点--服务治理

【架构】常见技术点--服务治理

导读&#xff1a;收集常见架构技术点&#xff0c;作为项目经理了解这些知识点以及解决具体场景是很有必要的。技术要服务业务&#xff0c;技术跟业务具体结合才能发挥技术的价值。 目录 1. 微服务 2. 服务发现 3. 流量削峰 4. 版本兼容 5. 过载保护 6. 服务熔断 7. 服务…
阅读更多...
微服务之流量控制

微服务之流量控制

Informal Essay By English I have been thinking about a question recently, what is the end of coding? 参考书籍&#xff1a; “凤凰架构” 流量控制 任何一个系统的运算、存储、网络资源都不是无限的&#xff0c;当系统资源不足以支撑外部超过预期的突发流量时&…
阅读更多...
数字信号处理8：利用Python进行数字信号处理基础

数字信号处理8：利用Python进行数字信号处理基础

我前两天买了本MATLAB信号处理&#xff0c;但是很无语&#xff0c;感觉自己对MATLAB的语法很陌生&#xff0c;看了半天也觉得自己写不出来&#xff0c;所以就对着MATLAB自己去写用Python进行的数字信号处理基础&#xff0c;我写了两天左右&#xff0c;基本上把matlab书上的代码…
阅读更多...
【数据结构】轻松掌握二叉树的基本操作及查找技巧

【数据结构】轻松掌握二叉树的基本操作及查找技巧

二叉树的基本操作 ​ 在学习二叉树的基本操作前&#xff0c;需先要创建一棵二叉树&#xff0c;然后才能学习其相关的基本操作。由于现在大家对二 叉树结构掌握还不够深入&#xff0c;为了降低学习成本&#xff0c;此处手动快速创建一棵简单的二叉树&#xff0c;快速进入二叉树操…
阅读更多...
【自然语言处理】不同策略的主题建模方法比较

【自然语言处理】不同策略的主题建模方法比较

不同策略的主题建模方法比较 本文将介绍利用 LSA、pLSA、LDA、NMF、BERTopic、Top2Vec 这六种策略进行主题建模之间的比较。 1.简介 在自然语言处理&#xff08;NLP&#xff09;中&#xff0c;主题建模一词包含了一系列的统计和深度学习技术&#xff0c;用于寻找文档集中的隐…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023