高性能的异步处理框架Disruptor(三)——Disruptor消费者的依赖关系

电玩女神 2023-06-06 03:47 288阅读 0赞

前文解释了Disruptor的生产者，消费者的简单工作流程。接下来学习当Disruptor的多个消费者存在依赖关系时，Disruptor对RingBuffer中的消息如何按照消费者的依赖关系来处理。

Disruptor系统的最初设计是为了支持需要按照特定的顺序发生的阶段性类似流水线事件，这种需求在企业应用系统开发中很常见。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70]  
如上图所示，对于独立的一个生产者和三个消费者。一个比较常见的依赖结构是: 消费者3必须等待前两个消费者处理完成后，才能开始工作。例如：消费者 C3 是具体业务逻辑。消费者 C1 可能在备份数据，而消费者 C2 可能在记录日志。

对于上述的业务场景，如果使用传统的消息队列，则至少需要四个队列：  
p1到c1的队列，p1到c2的队列，c1到c3的队列，c2到c3的队列。每个队列在消息进入队列和取出队列时都会产生消耗成本，因为传统队列在这种场景下效率并不高。

**在 Disruptor 的世界里，一切都由一个单独的 RingBuffer 管理:**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 1]  
基于我们前文对Disruptor向RingBuffer的读取写入策略，这张图的大部分逻辑都能理解——除了C3所依赖的ConsumerBarrier对象，别急我会解释清楚的。

先看上图的生产者模型，可以看到生产者的ProducerBarrier对象只维护了消费者C3，也就意外着，生产者来判断某个节点是否可写入，是否被消费者消费仅仅是基于消费者C3来的（如果消费者 C3 处理完了 Ring Buffer 的某一个节点，那么另外两个消费者肯定也处理完了）

此时管理消费者的依赖关系需要两个 ConsumerBarrier 对象。第一个仅仅与 RingBuffer 交互，C1 和 C2 消费者向它申请下一个可访问节点。第二个 ConsumerBarrier 只知道消费者 C1 和 C2，C3消费者向它申请两个消费者访问过的消息序号中较小的那个。

需要注意到是：第一个 ConsumerBarrier(CB1)用于告诉 C1 和 C2 消费者可以去访问RingBuffer中当前写入的最大序号的所有前面的数据。  
而第二个ConsumerBarrier (CB2) 不但会检查 RingBuffer 的序号，也会检查另外两个消费者的序号并且返回它们之间的最小值。因此，C3被告知可以访问 RingBuffer 里该序号前面的数据。

**C1，C2，C3 消费者还是直接从 Ring Buffer 拿数据节点——并不是由 C1 和 C2 消费者把数据节点从 Ring Buffer 里取出再传递给 C3 消费者的。是由第二个 ConsumerBarrier 告诉 C3 消费者，在 RingBuffer 里的哪些节点可以安全的处理。**

这是你可能会有另一个疑问，如果C3消费者要处理的数据依赖C1，C2的处理结果怎么办？  
解决办法就是把处理结果写入 RingBuffer 数据节点 (Entry) 本身。这样，当 C3 消费者从 RingBuffer 取出节点时，它已经填充好了 C3 消费者工作需要的所有信息。这里真正重要的地方是节点 (Entry) 对象的每一个字段应该只允许一个消费者写入。这可以避免产生并发写入冲突 (write-contention) 减慢了整个处理过程。例如：在节点Entry类中，定义两个字段：

public class RingBufferEntry{
       //只允许C1消费者写入的字段
       private Object consumerOneRes;
       //只允许C2消费者写入的字段
       private Object consumerTwoRes;
    }

如果消费者是C1, 它只写入字段 consumerOneRes。如果是C2, 它只写入字段 consumerTwoRes。第三个消费者C3，它只去读这两个字段但是不会做写入。

这一切看起来都要比队列实现更复杂。因为它涉及到更多的内部协调。但是这些细节对于消费者和生产者是透明的，它们只和 Barrier 对象交互。在消费者结构里。上文例子中提到的菱形结构可以用下面的方法创建:

//创建C1，C2
    ConsumerBarrier consumerBarrier1 = ringBuffer.createConsumerBarrier(); 
    BatchConsumer consumer1 = new BatchConsumer(consumerBarrier1, handler1); 
    BatchConsumer consumer2 = new BatchConsumer(consumerBarrier1, handler2); 
    //创建依赖C1，C2的C3
    ConsumerBarrier consumerBarrier2 = ringBuffer.createConsumerBarrier(consumer1, consumer2); 
    BatchConsumer consumer3 = new BatchConsumer(consumerBarrier2, handler3); 
    //生产者依赖C3来确定完成消费的Entry
    ProducerBarrier producerBarrier = ringBuffer.createProducerBarrier(consumer3);

如何关联 Disruptor 与相互依赖(等待)的多个消费者。重点是:

>  *  使用多个 ConsumerBarrier 来管理消费者之间的依赖(等待)关系。
>  *  使用 ProducerBarrier 监视结构图中最后一个消费者。
>  *  只允许一个消费者更新数据节点 (Entry) 的每一个独立字段。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70]: /images/20230601/33719ede74d64eb5bdb2855e57f78ea9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 1]: /images/20230601/9e538b7b9e554255abcb3cf3b3e98bd4.png