Dubbo订阅发布的实现原理

Dubbo的订阅分布实现可基于注册中心，也可简单的通过直接通信的方式实现，但直接通信的方式可靠性比较低，目前大多数的实现还是需要一个注册中心。Dubbo内部实现了几个基于不同组件实现的注册中心(Redis,Nacos,Zookeeper等等)，因为主要是了解订阅发布实现的原理，所以我基于最较简单的Redis为入口进行了解学习。

整体流程

先简单梳理下整体流程

1. 服务治理中心（dubbo-admin）启动，并同时订阅所有消费者、服务提供者、路由和配置元数据信息。
2. provider启动时，会向注册中心写入自己的元数据信息，同时会订阅配置元数据信息。
3. consumer启动时，也会向注册中心写入自己的元数据信息，并订阅provider、路由和配置元数据信息。
4. 当有provider离开或有新的provider加入时，注册中心provider的目录会发生变化，变化信息会动态通知给消费者，服务治理中心。

发布的实现

1. 采用Redis的过期机制和publish/subscribe特性实现发布
2. provider发布服务时，会在Redis中创建一个kv（默认超时时间为1分钟），并且在redis通道中发布一条register事件消息。
3. provider实例化的同时，也会实例化一个RedisRegistry，RedisRegistry的构造方法中会启动一个expireExecutor定时调度线程池，该定时任务会不停地向Redis发送请求并将当前provider在redis中的kv进行延时
4. 当provider的kv对已经过期，则将对应的kv从redis删除，并向redis通道发布一条unregister事件消息。

伪代码如下（具体代码可见2.6.x及以下的源码）：

public class RedisRegistry{
   
 
    // 定时调度线程池, 不断地调用deferExpired()
    private final ScheduledExecutorService expireExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryExpireTimer", true));

    private final ScheduledFuture<?> expireFuture;

  // 注册实例初始化，对于每一个dubbo角色（provider/consumer/admin）其都会有一个register实例，因为每一个dubbo角色都会需要发布/订阅的功能
    public RedisRegistry() {
   
        // 一些初始化操作
        // ...
    
        this.expireFuture = expireExecutor.scheduleWithFixedDelay(new Runnable() {
   
            @Overrider
            public void run() {
   
                try{
   
                   deferExpired();
                }catch(
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22