disruptor源码分析一之核心组件介绍

山行AI

发布于 2019-07-01 11:04:58

1.2K0

发布于 2019-07-01 11:04:58

文章被收录于专栏：山行AI

disruptor是LAMX用于交易场景的一个环形队列。按照disruptor的官方wiki中说的，学习disruptor的最好的方式就是与java中的BlockingQueue进行比较。在disruptor中同一个进程中的线程间数据的移动是依托于 messages或者events。和queue相同的一些关键特性中，disruptor提供了更好的实现，比如：

消费者的事件多投和消费者依赖图;
事件的内存预分配;
无锁操作。

核心组件

Ring Buffer:环形缓冲区，之前被认为是disruptor的主要组件，在3.0版本以后仅仅作为disruptor中存储和更新events（数据）的介质。

Sequence: 用于标识特殊组件的位置。每个consumer(EventProcessor)和disruptor一样也自己维持一个Sequence。主要的并发代码依赖于Sequence中值的改变，所以Sequence支持很多AtomicLong当前的很多特性。事实上两者唯一的不同是Sequence包括防止多个Sequence和其他值之间伪共享的特性。

Sequencer: 是disruptor的真正核心。这个接口的两种实现(single producer和 multi producer)实现了所有的并发算法，用来更快更正确地在生产者与消费者之间传递数据。

Sequence Barrier: 是由Sequencer创建，消费者与缓存RingBuffer之间的桥梁。它包含Sequencer发布的Sequence的引用和每个消费者依赖的Sequences的引用。它一般情况下用来判断是否有可用的events(数据)供消费者消费。

Wait Strategy: 等待策略。用来决定消费者如何等待消费被生产者放进disruptor中的events(数据)。对实现lock-free也有一定的帮助。
Event: 是生产者与消费者之间数据传递的数据单元。一般都是由用户自定义的。

EventProcessor: 事件执行器，是消费者线程池Executor的调度单元。是disruptor处理events(数据)的主要的event loop，对consumer的sequence拥有主导权。BatchEventProcessor是其中的一个代表，它包括一个有效的event loop的实现，并且会回调用户实现的EventHandler接口。

EventHandler: 事件处理器。需要用户自己实现的接口，涉及到用户自己的业务逻辑，代表disruptor的一个消费者。EventProcessor是在EventHandler的基础上的一层封装。
Producer: 用户自定的代码，用来将events(数据)从disruptor中出列。

有多个生产者和多个消费者的disruptor效果图：

注意观察图中的JournalConsumer,ReplicationConsumer与ApplicationConsumer之间的关系。

disruptor的特性

1. 事件多投(Multicast Events)

这是普通队列和disruptor之间行为上最大的一个不同点。当你有多个消费者监听同一个disruptor时，所有的events(数据)被发送到所有的消费者中，但是普通队列与此相反，一个event只会被发送到一个消费者。disruptor的这个行为主要用于你需要对同一个数据并行地做不同的处理的场景。在LAMX中的应用实例是，当需要同时进行三个操作时，这三个操作分别是：

journalling: 把输入的数据持久化到journal文件中，对应上图中的JournalConsumer;
replication: 把输入的数据发送到另一台机器中备份，对应上图中的ReplicationConsumer;
business logic: 业务逻辑处理,对应上图中的ApplicationConsumer。用执行器(Executor)形式的事件处理，可能会使用到disruptor中的WorkerPool，来并行的处理不同的events。这个是建立在已经存在的disruptor基础上做的，并不是作为相同的第一级支持对待，所以在达到和它相同的目录时，它可能并不是最有效的方式。

2. 消费者依赖图（Consumer Dependency Graph）

为了支持真实使用中的并行处理行为，所以需要支持多个消费者之间的协作。反过头来看一下上面描述的例子，我们需要让业务逻辑处理的消费者ApplicationConsumer等待journalConsumer和ReplicationConsumer 完成他们的任务之后再运行。我们把这部分称为闸，更准确的说法是这种行为的超级称为闸。闸主要出现在两个地方。首先我们需要确保生产者的生产能力不能超出消费者太多，这个通过调用RingBuffer.addGatingConsumers() 添加相关的消费者到Disruptor中来处理。还有一点就是，前面提到的场景是通过构造一个SequenceBarrier,这个SequenceBarrier中包含着很多个Sequence，这些Sequence必须先完成它们对应的操作。

上面提到的三个消费者同时监听Ring Buffer中的Events.这个例子是有一个依赖图的。ApplicationConsumer依赖于JournalConsumer和ReplicationConsumer。这意味着JournalConsumer和ReplicationConsumer 可以不受约束的相互之间并行运行。这种依赖关系可以通过ApplicationConsumer的SequenceBarrier与JournalConsumer和ReplicationConsumer的Sequences之间的联系来理解。Sequencer与它下游的消费者之间的关系也是值得好好说一说的。它的角色之一是用来确保生产者发布的events不要包裹住整个RingBuffer环。为了达到这个，下游消费者的Sequence要小于RingBuffer的Sequence，而RingBuffer的Sequence大小要小于RingBuffer的大小。使用依赖图是一件很有趣的优化操作。因为ApplicationConsumer的Sequence保证要小于或等于JournalConsumer和ReplicationConsumer的Sequence(这就是依赖关系所需要确保的)，所以Sequencer仅仅只需要关注ApplicationConsumer的Sequence变化。在更常见的情况下，Sequencer只需要知道消费者们的所有Sequence都是整个依赖树的叶子节点。

3. 事件的内存预分配(Event Preallocation)

disruptor的一个目标是提供一个低延迟的环境。在低延迟的系统中需要减少或消除内存的分配。在java系统中这么做的目的主要是为了减少gc。在低延迟的c/c++系统中，频繁的内存分配也会是一个问题，因为在内存分配器的竞争使用上会频繁。

为了支持低延迟，disruptor提供了内存预分配。在构造时由用户提供的EventFactory会被Disruptor的RingBuffer的每个entry中被调用。当发布新的数据到disruptor中去时，api会允许用户持有构造过的object以便于它们能调用保存的这些object中的方法或者更新里面的字段。disruptor保证了这些操作的线程安全，只要它们被正确的实现了。

4. 无锁(Optionally Lock-free)

另外一个低延迟的关键点是disruptor中的无锁算法拓展实现。所有的内存可见性和正确性的保证都是通过使用内存屏障和CAS操作实现的。唯一的一个使用了真实的锁的地方就是BlockingWaitStrategy。这个单独这样做的目的是使用了一个Condition对象来让一个消费线程在等待新的events(数据)时会被阻塞住。大多数的低延迟系统会使用忙时等待来避免使用Condition可能引起的抖动，但是很大一部分系统的忙时等待操作会导致系统性能的下降，尤其是cpu资源的严重占用。比如虚拟环境下的web服务器。