天天看点
返回

[从源码学设计]蚂蚁金服SOFARegistry之消息总线

SOFARegistry 是蚂蚁金服开源的一个生产级、高时效、高可用的服务注册中心。本系列文章重点在于分析设计和架构，即利用多篇文章，从多个角度反推总结 DataServer 或者 SOFARegistry 的实现机制和架构思路，让大家借以学习阿里如何设计。本文为第四篇，介绍SOFARegistry之消息总线。

[从源码学设计]蚂蚁金服SOFARegistry之消息总线

目录

    • 3.1 目录结构
    • 3.2 类定义
    • 3.3 Listener
    • 3.4 核心消息
    • 3.5 主要逻辑
    • 3.2.1 操作
    • 3.2.2 执行 & 解耦
    • 3.2.3 Listener列表
    • 3.2.4 ArrayListMultimap
    • 3.3.1 基类
    • 3.3.2 派生类
    • 3.3.2 自动注册
    • 2.1 业务范畴
    • 2.2 问题点
    • 2.3 解决方案
    • 1.1 事件驱动模型
    • 1.2 消息总线
    • 1.1.1 概念
    • 1.1.2 应用环境
    • 0x00 摘要
    • 0x01 相关概念
    • 0x02 业务领域
    • 0x03 EventCenter
    • 0x04 总结
    • 0xFF 参考

SOFARegistry 是蚂蚁金服开源的一个生产级、高时效、高可用的服务注册中心。

本系列文章重点在于分析设计和架构，即利用多篇文章，从多个角度反推总结 DataServer 或者 SOFARegistry 的实现机制和架构思路，让大家借以学习阿里如何设计。

本文为第四篇，介绍SOFARegistry之消息总线。

事件驱动模型，也即是我们通常说的观察者。基于发布-订阅模式的编程模型。

定义对象间的一种一对多的依赖关系，当一个对象的状态发生变化时，所有依赖它的对象都得到通知并自动更新。

从程序设计的角度来看，事件驱动模型的核心构件通常包含以下几个：

  • 事件源：负责产生事件的对象。比如我们常见的按钮，按钮就是一个事件源，能够产生“点击”这个事件
  • 事件监听器（事件处理器）：负责处理事件的对象
  • 事件：或者称为事件对象，是事件源和事件监听器之间的信息桥梁。是整个事件模型驱动的核心

当我们面对如下的环境时，事件驱动模型通常是一个好的选择：

  • 程序中有许多任务；
  • 任务之间高度独立（因此它们不需要互相通信，或者等待彼此）；
  • 在等待事件到来时，某些任务会阻塞；

总线（Bus）一般指计算机各种功能部件之间传送信息的公共通信干线，而EventBus则是事件源（publisher)向订阅方（subscriber)发送订阅事件的总线，它解耦了观察者模式中订阅方和事件源之间的强依赖关系。

消息总线扮演着一种消息路由的角色，拥有一套完备的路由机制来决定消息传输方向。发送端只需要向消息总线发出消息而不用管消息被如何转发，为了避免消息丢失，部分消息总线提供了一定的持久化存储和灾备的机制。

消息总线简单理解就是一个消息中心，众多微服务实例可以连接到总线上，实例可以往消息中心发送或接收信息（通过监听）。

一般的应用的场景就是在用观察者模式的地方就可以用EventBus进行替代。

[从源码学设计]蚂蚁金服SOFARegistry之消息总线

DataServer 本质上是一个网络应用程序，所以有如下特点：

  • 需要处理各个方面发送来的消息；
  • 程序中任务繁多，任务之间独立，大多数任务不存在互斥通讯等操作；在等待事件到来时，某些任务会阻塞；
  • 某一个消息往往有多个投递源；

因此天然适合用事件驱动机制来实现。

能够想到的问题点如下：

  • 因为一个事件往往会有多个投递源，如何解耦事件投递和事件处理之间的逻辑？
  • 怎样实现Listener一次注册，就能够知道Listener对那些事件感兴趣的，进而在有某类事件发生时通知到Listener的呢？
  • 如何使得一个Listener可以处理多个事件？
  • 如何使得一个事件被多个Listener处理？
  • 可否简化注册流程？
  • 是否需要维护消息顺序？
  • 处理消息方式是异步还是同步？
  • 多个同样消息是否要归并？

具体我们在后文会详述阿里的思路。

DataServer 内部逻辑主要是通过事件驱动机制来实现的，下图列举了部分事件在事件中心的交互流程，从图中可以看到，一个事件往往会有多个投递源，非常适合用 EventCenter 来解耦事件投递和事件处理之间的逻辑；

[从源码学设计]蚂蚁金服SOFARegistry之消息总线

业界消息总线有很多，比如 Android EventBus是一个发布/订阅事件总线框架，基于观察者模式，将事件的接收者和发送者分开，简化了组件之间的通信。

而SOFARegistry EventCenter 的作用也类似：从逻辑上解耦，将事件的接收者和发送者分开，简化组件之间通信。阿里的实现有自己的特点，开发者可以借鉴这里的使用技巧和思路。

├── event
│   ├── AfterWorkingProcess.java
│   ├── DataServerChangeEvent.java
│   ├── Event.java
│   ├── EventCenter.java
│   ├── LocalDataServerChangeEvent.java
│   ├── MetaServerChangeEvent.java
│   ├── RemoteDataServerChangeEvent.java
│   ├── StartTaskEvent.java
│   ├── StartTaskTypeEnum.java
│   └── handler
│       ├── AbstractEventHandler.java
│       ├── AfterWorkingProcessHandler.java
│       ├── DataServerChangeEventHandler.java
│       ├── LocalDataServerChangeEventHandler.java
│       ├── MetaServerChangeEventHandler.java
│       └── StartTaskEventHandler.java

类定义如下：

public class EventCenter {

    private Multimap<Class, AbstractEventHandler> MAP = ArrayListMultimap.create();

    /**
     * eventHandler register
     * @param handler
     */
    public void register(AbstractEventHandler handler) {
        List<Class> interests = handler.interest();
        for (Class interest : interests) {
            MAP.put(interest, handler);
        }
    }

    /**
     * event handler handle process
     * @param event
     */
    public void post(Event event) {
        Class clazz = event.getClass();
        if (MAP.containsKey(clazz)) {
            Collectionhandlers = MAP.get(clazz);
            if (handlers != null) {
                for (AbstractEventHandler handler : handlers) {
                    handler.handle(event);
                }
            }
        } else {
            throw new RuntimeException("no suitable handler was found:" + clazz);
        }
    }
}

普通 EventBus 大多有三个操作：

  • 注册 Listener--register (Object Listener)；
  • 注销 Listener--unregister (Object Listener)；
  • 发布 Event--post (Object event)；

但是阿里的EventCenter并没有注销操作，因为业务上不需要，所以只有如下接口。

  • register(AbstractEventHandler handler) 的工作就是找出这个Listener对哪些事件感兴趣，然后把这种事件类型和对应的Listener注册到 EventCenter；
  • 当post一个event时候，会遍历这个消息的处理函数列表，逐一调用处理函数，其实就是同步执行了，当然也许 EventHandler 内部自己实现了异步；因为是同步执行，所以不需要维持消息的有序性，否则需要使用queue来实现每个线程post的Event是有序的；

具体使用举例如下：在MetaServerChangeEventHandler中有如下代码投放消息。

eventCenter.post(new StartTaskEvent(set));

eventCenter.post(new DataServerChangeEvent(result.getNodes(), versionMap,
        DataServerChangeEvent.FromType.REGISTER_META));

handler中声明了自己支持什么种类的event，当register时候，会以event为key，把自己注册到eventCenter的map中，在 post 函数中，根据event的class，取出了handler，从而执行，也做到了解耦。

在观察者模式中，事件源中会维护一个Listener的列表，而且向这个事件源注册的Listener一般只会收到一类事件的通知，如果Listener对多个不同类的事件感兴趣，则需要向多个事件源注册。

EventCenter 是怎样实现Listener一次注册，能够知道Listener对那些事件感兴趣的，进而在有某类事件发生时通知到Listener的呢？

答案在ArrayListMultimap，其key是Event，其 Value 就是 AbstractEventHandler。这个 map 就是 Event 事件类型 和对其感兴趣的处理函数的列表，一个 Event 可能有多个处理函数。

顾名思义，

com.google.common.collect.ArrayListMultimap

可以在key对应的value中设置一个ArrayList。这样就保证了一个事件可以有多个处理函数。

具体可以见下例子。

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;

import java.util.Collection;

public class testArrayListMultimap {
     static void main() {
        Multimapmultimap = ArrayListMultimap.create();
        multimap.put("fruit", "banana");
        multimap.put("fruit", "apple");
        multimap.put("fruit", "apple");
        multimap.put("fruit", "peach");
        multimap.put("fish","crucian");
        multimap.put("fish","carp");

        System.err.println(multimap.size());//6
        Collectionfruits = multimap.get("fruit");
        System.err.println(fruits);//[bannana, apple, apple, peach]
    }
}

Listener 是由 AbstractEventHandler 的派生类实现的。

EventHandler基类AbstractEventHandler定义具体如下：

public abstract class AbstractEventHandlerimplements InitializingBean {

    @Autowired
    private EventCenter         eventCenter;

    @Override
    public void afterPropertiesSet() throws Exception {
        eventCenter.register(this);
    }

    /**
     * event handle func
     * @param event
     */
    public void handle(Event event) {
            doHandle(event);
    }

    public abstract List<Class> interest();

    public abstract void doHandle(Event event);
}

其主要作用为三点：

  • 派生类必须实现interest来声明自己想处理什么Event，而且Event是配置在一个数组中，这样就使得一个函数可以处理多个事件。 
@Override
public List<Class> interest() {
    return Lists.newArrayList(LocalDataServerChangeEvent.class);
}
  • 派生类实现doHandle来处理消息；
  • 因为afterPropertiesSet中做了设定，所以每一个继承此类的Handler都会自动注册到EventCenter之中。

以MetaServerChangeEventHandler为例，只要在interest函数中声明自己对哪些消息感兴趣，在doHandle函数中实现业务即可。

public class MetaServerChangeEventHandler extends AbstractEventHandler{
  
    @Override
    public List<Class> interest() {
        return Lists.newArrayList(MetaServerChangeEvent.class);
    }
  
		@Override
    public void doHandle(MetaServerChangeEvent event) {
       ......
    }
}

这里需要专门说一下自动注册，因为初接触者很容易疏漏从而感到奇怪。

自动注册使用的是Spring的afterPropertiesSet方法完成。

afterPropertiesSet方法可以针对某个具体的bean进行配置，其将在Bean所有的属性被初始化后调用，但是会在init前调用。afterPropertiesSet 必须实现 InitializingBean接口。

package org.springframework.beans.factory;

public interface InitializingBean {
    void afterPropertiesSet() throws Exception;
}

基类AbstractEventHandler实现InitializingBean接口。

public abstract class AbstractEventHandlerimplements InitializingBean

而每一个派生类就注册了派生类本身到eventCenter。

@Override
public void afterPropertiesSet() throws Exception {
    eventCenter.register(this);
}

具体涉及到业务，EventCenter主要处理三种消息：

  • DataServerChangeEvent，是其他Data Server的节点变化消息；
  • MetaServerChangeEvent，是Meta Sever的变化消息；
  • StartTaskEvent：；

分别对应三个消息处理handler：

  • public class DataServerChangeEventHandler extends AbstractEventHandler
  • public class MetaServerChangeEventHandler extends AbstractEventHandler
  • public class StartTaskEventHandler extends AbstractEventHandler

我们用 StartTaskEvent 举例，具体消息内容根据具体业务设置。

public class StartTaskEvent implements Event {
    private final SetsuitableTypes;

    public StartTaskEvent(SetsuitableTypes) {
        this.suitableTypes = suitableTypes;
    }

    public SetgetSuitableTypes() {
        return suitableTypes;
    }
}

EventCenter主要逻辑如下图所示：

          +------------------------------+
          | MetaServerChangeEventHandler |
          +-----------+------------------+
                      |
                      |  post(new StartTaskEvent)
                      |
                      |
                      |                                      +------------------------+
                      v                                      |  StartTaskEventHandler |
+---------------------+-----------------------+              |                        |
|                EventCenter                  |              | +--------------------+ |
|                                             |              | |                    | |
| +-----------------------------------------+ +---------------------> doHandle      | |
| |Multimap<, AbstractEventHandler> | |              | |                    | |
| +-----------------------------------------+ | <--------------+ afterPropertiesSet | |
|                                             |  register    | |                    | |
+---------------------------------------------+              | |      interest      | |
                                                             | |                    | |
                                                             | +--------------------+ |
                                                             +------------------------+

手机如下图：

[从源码学设计]蚂蚁金服SOFARegistry之消息总线

SOFARegistry EventCenter 的作用与业界大多总线类似：从逻辑上解耦，将事件的接收者和发送者分开，简化组件之间通信。但是阿里的实现有自己的特点，开发者可以借鉴这里的使用技巧和思路。

针对我们前面提出的问题，现在回答如下：

    • 答案：handler中声明了自己支持什么种类的event，当register时候会以event为key，把自己注册到eventCenter的map中；在 post 函数中，根据event的class，取出了handler从而执行，也做到了解耦。
    • 答案：派生类必须实现interest来声明自己想处理什么Event；
    • 答案：接上问题，Event是配置在一个数组中，这样就使得一个函数可以处理多个事件。
    • 答案：采用ArrayListMultimap实现listener列表；
    • 答案：自动注册，派生类不需要操心。afterPropertiesSet中做了设定，所以每一个继承此类的Handler都会自动注册到EventCenter之中。
    • 答案：不需要，因为是同步处理；
    • 答案：这里是同步；
    • 答案：这里不需要归并，没有业务需求；

Guava中EventBus分析

蚂蚁金服服务注册中心如何实现 DataServer 平滑扩缩容

消息总线 SOFARegistry
上一篇: 第二次站立会议10
下一篇: 性能测试工具JMeter 基础（六）—— 测试元件： 线程组

继续阅读