天天看点

[从源码学设计]蚂蚁金服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 平滑扩缩容

继续阅读