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

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

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

导入的依赖：

    4.0.0

    com.jxz
    NacosListenerTest
    1.0-SNAPSHOT

    
        org.springframework.boot
        spring-boot-starter-parent
        2.3.12.RELEASE
    

    
        1.8
        Hoxton.SR12
        2.2.8.RELEASE
    

    
        
            org.springframework.boot
            spring-boot-starter-web
        

        
            com.alibaba.cloud
            spring-cloud-starter-alibaba-nacos-discovery
        

        
            com.alibaba.cloud
            spring-cloud-starter-alibaba-nacos-config
        

        
            org.projectlombok
            lombok
            1.18.24
        
    

    
        
            
            
                org.springframework.cloud
                spring-cloud-dependencies
                ${spring-cloud.version}
                pom
                import
            
            
            
                com.alibaba.cloud
                spring-cloud-alibaba-dependencies
                ${spring-cloud-alibaba.version}
                pom
                import
            
        
    


    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
            
        
    



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

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
2
3
4
5
6
7
8
9
10
11
12

2. 实现一个订阅者来处理发布的事件，需要继承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());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

这里有三个方法：

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

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

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

3. 起一个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());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

打印结果如下，显然是按照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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

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

定位到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()));
}
1
2
3
4
5
6
7
8
9
10

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

@Override
public void run() {
    openEventHandler();
}
1
2
3
4

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

public class DefaultPublisher extends Thread implements EventPublisher {}
1

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

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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

这里其实就是观察者模式的一大应用，参考我之前写的一篇帮助理解的文章->简单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);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

通过上面的分析我们可以得到，事件发布是通过 阻塞队列+异步任务的方式来实现的，订阅者执行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;
    }
    
    // 这里代码省略
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

上面就用到了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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

通过debug看以下内容：

MapUtil.computeIfAbsent完成的动作是，对每一个Event事件，创建一个对应的EventPublisher对象，这个对象中包含着订阅者信息，也就是一个ConcurrentHashSet 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;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

首先有一个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);
}
1
2
3
4
5
6
7

然后就是用一个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);
    }
};
1
2
3
4
5
6
7
8
9
10

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

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

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