从零开始实现lmax-Disruptor队列（二）多消费者、消费者组间消费依赖原理解析

技术分享 3年前 (2022-06-10) 0 999+

关注

MyDisruptor V2版本介绍

在v1版本的MyDisruptor实现单生产者、单消费者功能后。按照计划，v2版本的MyDisruptor需要支持多消费者和允许设置消费者组间的依赖关系。

由于该文属于系列博客的一部分，需要先对之前的博客内容有所了解才能更好地理解本篇博客

v1版本博客：从零开始实现lmax-Disruptor队列（一）RingBuffer与单生产者、单消费者工作原理解析

MyDisruptor支持多消费者

disruptor中的生产者和消费者是互相制约的，生产者的生产速度不能过快，在逻辑上队列已满时需要阻塞等待消费者进行消费，直到队列不满。
而要支持多消费者，上述描述需要做一定的调整：即生产者的生产速度不能过快，在逻辑上队列已满时需要阻塞等待”最慢的消费者“完成消费，直到队列不满。
disruptor中每个消费者都拥有自己的消费序列号，生产者在生产时需要保证生产的序列号不能覆盖任何一个消费者，即生产者的序列号不能超过最慢的消费者序列号一圈（Producer.Sequence - SlowestConsumer.Sequence <= ringBufferSize）
所以生产者需要维护一个消费者序列集合，在生产者生产时控制生产速度避免超过最慢的消费者而发生越界。

v2版本单线程生产者实现

package mydisruptor;  import mydisruptor.util.SequenceUtil; import mydisruptor.waitstrategy.MyWaitStrategy;  import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.locks.LockSupport;  /**  * 单线程生产者序列器（仿Disruptor.SingleProducerSequencer）  * 只支持单消费者的简易版本（只有一个consumerSequence）  *  * 因为是单线程序列器，因此在设计上就是线程不安全的  * */ public class MySingleProducerSequencer {      /**      * 生产者序列器所属ringBuffer的大小      * */     private final int ringBufferSize;      /**      * 当前已发布的生产者序列号      * （区别于nextValue）      * */     private final MySequence currentProducerSequence = new MySequence();      /**      * 生产者序列器所属ringBuffer的消费者序列集合      * （v2版本简单起见，先不和disruptor一样用数组+unsafe来实现）      * */     private final List<MySequence> gatingConsumerSequenceList = new ArrayList<>();      private final MyWaitStrategy myWaitStrategy;      /**      * 当前已申请的序列(但是是否发布了，要看currentProducerSequence)      *      * 单线程生产者内部使用，所以就是普通的long，不考虑并发      * */     private long nextValue = -1;      /**      * 当前已缓存的消费者序列      *      * 单线程生产者内部使用，所以就是普通的long，不考虑并发      * */     private long cachedConsumerSequenceValue = -1;      public MySingleProducerSequencer(int ringBufferSize, MyWaitStrategy myWaitStrategy) {         this.ringBufferSize = ringBufferSize;         this.myWaitStrategy = myWaitStrategy;     }      /**      * 一次性申请可用的1个生产者序列号      * */     public long next(){         return next(1);     }      /**      * 一次性申请可用的n个生产者序列号      * */     public long next(int n){         // 申请的下一个生产者位点         long nextProducerSequence = this.nextValue + n;         // 新申请的位点下，生产者恰好超过消费者一圈的环绕临界点序列         long wrapPoint = nextProducerSequence - this.ringBufferSize;          // 获得当前已缓存的消费者位点         long cachedGatingSequence = this.cachedConsumerSequenceValue;          // 消费者位点cachedValue并不是实时获取的（因为在没有超过环绕点一圈时，生产者是可以放心生产的）         // 每次发布都实时获取反而会触发对消费者sequence强一致的读，迫使消费者线程所在的CPU刷新缓存（而这是不需要的）         if(wrapPoint > cachedGatingSequence){             // 比起disruptor省略了if中的cachedGatingSequence > nextProducerSequence逻辑             // 原因请见：https://github.com/LMAX-Exchange/disruptor/issues/76              // 比起disruptor省略了currentProducerSequence.set(nextProducerSequence);             // 原因请见：https://github.com/LMAX-Exchange/disruptor/issues/291             long minSequence;              // 当生产者发现确实当前已经超过了一圈，则必须去读最新的消费者序列了，看看消费者的消费进度是否推进了             // 这里的getMinimumSequence方法中是对volatile变量的读，是实时的、强一致的读             while(wrapPoint > (minSequence = SequenceUtil.getMinimumSequence(nextProducerSequence, gatingConsumerSequenceList))){                 // 如果确实超过了一圈，则生产者无法获取可用的队列空间，循环的间歇性park阻塞                 LockSupport.parkNanos(1L);             }              // 满足条件了，则缓存获得最新的消费者序列             // 因为不是实时获取消费者序列，可能cachedValue比上一次的值要大很多             // 这种情况下，待到下一次next申请时就可以不用去强一致的读consumerSequence了             this.cachedConsumerSequenceValue = minSequence;         }          // 记录本次申请后的，已申请的生产者位点         this.nextValue = nextProducerSequence;          return nextProducerSequence;     }      public void publish(long publishIndex){         // 发布时，更新生产者队列         // lazySet，由于消费者可以批量的拉取数据，所以不必每次发布时都volatile的更新，允许消费者晚一点感知到，这样性能会更好         // 设置写屏障         this.currentProducerSequence.lazySet(publishIndex);          // 发布完成后，唤醒可能阻塞等待的消费者线程         this.myWaitStrategy.signalWhenBlocking();     }      public MySequenceBarrier newBarrier(){         return new MySequenceBarrier(this.currentProducerSequence,this.myWaitStrategy,new ArrayList<>());     }      public MySequenceBarrier newBarrier(MySequence... dependenceSequences){         return new MySequenceBarrier(this.currentProducerSequence,this.myWaitStrategy,new ArrayList<>(Arrays.asList(dependenceSequences)));     }      public void addGatingConsumerSequenceList(MySequence newGatingConsumerSequence){         this.gatingConsumerSequenceList.add(newGatingConsumerSequence);     }      public int getRingBufferSize() {         return ringBufferSize;     } }  /**  * 序列号工具类  * */ public class SequenceUtil {      /**      * 从依赖的序列集合dependentSequence和申请的最小序列号minimumSequence中获得最小的序列号      * @param minimumSequence 申请的最小序列号      * @param dependentSequenceList 依赖的序列集合      * */     public static long getMinimumSequence(long minimumSequence, List<MySequence> dependentSequenceList){         for (MySequence sequence : dependentSequenceList) {             long value = sequence.get();             minimumSequence = Math.min(minimumSequence, value);         }          return minimumSequence;     }      /**      * 获得传入的序列集合中最小的一个序列号      * @param dependentSequenceList 依赖的序列集合      * */     public static long getMinimumSequence(List<MySequence> dependentSequenceList){         // Long.MAX_VALUE作为上界，即使dependentSequenceList为空，也会返回一个Long.MAX_VALUE作为最小序列号         return getMinimumSequence(Long.MAX_VALUE, dependentSequenceList);     } }

v2版本生产者相对于v1版本的一个变化就是将维护的单一消费者序列consumerSequence变为了一个容纳多消费者序列的集合gatingConsumerSequenceList，并提供了动态新增消费者序列的接口addGatingConsumerSequenceList方法。
在申请可用生产序列号的方法next中，判断是否越界的条件也由v1版本的wrapPoint > consumerSequence,变成了wrapPoint > SequenceUtil.getMinimumSequence()。
SequenceUtil.getMinimumSequence方法接收一个序列号集合和一个生产者序列号，返回其中的最小序列值。如果环绕越界点序列大于了返回的最小序列值，则说明所要申请的序列号已经越界了（快于最慢消费者一圈），需要等待最慢的消费者消费，令生产者阻塞。

MyDisruptor支持消费者组间消费依赖

v2版本中除了要支持多消费者，还需要支持消费者的组间消费依赖，例如有三个消费者A、B、C，消费依赖关系为A/B -> C, 即对于生产者发布的任意一个事件在AB消费成功后C才能进行消费（A，B之间的消费顺序不限制）。
消费者的组间消费依赖关系可以很复杂(但不能存在循环依赖）。
要实现消费者间的依赖，关键思路是让每个消费者维护其上游消费者的序列，在消费时控制所消费的序列号不大于上游所依赖的最慢的消费者。

/**  * 序列栅栏（仿Disruptor.SequenceBarrier）  * */ public class MySequenceBarrier {      private final MySequence currentProducerSequence;     private final MyWaitStrategy myWaitStrategy;     private final List<MySequence> dependentSequencesList;      public MySequenceBarrier(MySequence currentProducerSequence,                              MyWaitStrategy myWaitStrategy, List<MySequence> dependentSequencesList) {         this.currentProducerSequence = currentProducerSequence;         this.myWaitStrategy = myWaitStrategy;         this.dependentSequencesList = dependentSequencesList;     }      /**      * 获得可用的消费者下标      * */     public long getAvailableConsumeSequence(long currentConsumeSequence) throws InterruptedException {         // v1版本只是简单的调用waitFor，等待其返回即可         return this.myWaitStrategy.waitFor(currentConsumeSequence,currentProducerSequence,dependentSequencesList);     } }  /**  * 阻塞等待策略  * */ public class MyBlockingWaitStrategy implements MyWaitStrategy{      private final Lock lock = new ReentrantLock();     private final Condition processorNotifyCondition = lock.newCondition();      @Override     public long waitFor(long currentConsumeSequence, MySequence currentProducerSequence, List<MySequence> dependentSequences)             throws InterruptedException {         // 强一致的读生产者序列号         if (currentProducerSequence.get() < currentConsumeSequence) {             // 如果ringBuffer的生产者下标小于当前消费者所需的下标，说明目前消费者消费速度大于生产者生产速度              lock.lock();             try {                 //                 while (currentProducerSequence.get() < currentConsumeSequence) {                     // 消费者的消费速度比生产者的生产速度快，阻塞等待                     processorNotifyCondition.await();                 }             }             finally {                 lock.unlock();             }         }          // 跳出了上面的循环，说明生产者序列已经超过了当前所要消费的位点（currentProducerSequence > currentConsumeSequence）         long availableSequence;         if(!dependentSequences.isEmpty()){             // 受制于屏障中的dependentSequences，用来控制当前消费者消费进度不得超过其所依赖的链路上游的消费者进度             while ((availableSequence = SequenceUtil.getMinimumSequence(dependentSequences)) < currentConsumeSequence) {                 // 由于消费者消费速度一般会很快，所以这里使用自旋阻塞来等待上游消费者进度推进（响应及时，且实现简单）                  // 在jdk9开始引入的Thread.onSpinWait方法，优化自旋性能                 MyThreadHints.onSpinWait();             }         }else{             // 并不存在依赖的上游消费者，大于当前消费进度的生产者序列就是可用的消费序列             availableSequence = currentProducerSequence.get();         }          return availableSequence;     }      @Override     public void signalWhenBlocking() {         lock.lock();         try {             // signal唤醒所有阻塞在条件变量上的消费者线程（后续支持多消费者时，会改为signalAll）             processorNotifyCondition.signal();         }         finally {             lock.unlock();         }     } }

v2版本中，控制消费者消费速度的组件,MySequenceBarrier序列屏障中除了需要维护生产者的序列号，避免消费越界外，还新增了一个List类型的成员变量dependentSequencesList用于维护所依赖的上游的一至多个消费者的序列号对象。
获取可用的最大消费者序列号方法MyWaitStrategy.waitFor方法也新增参数传入依赖的上游消费者序列集合，用于控制返回的最大可消费序列号不会大于上游最慢的消费者序列。
在阻塞等待策略MyBlockingWaitStrategy中，和等待生产者生产时的策略不同，等待上游最慢消费者消费时并不是通过Condition.await方法令线程陷入阻塞态，而是while自旋等待。
这是因为生产者地下一次生产是不可预知的，有可能会长时间等待；比起自旋，阻塞等待可以极大地减少对CPU资源的消耗。而上游消费者的消费速度则一般很快，生产者生产到上游消费者消费完之间的时间间隔会很短，所以使用自旋来实现消费者间的消费依赖
注意：正是因为在消费者的实现中假设了上游消费者的“理论”消费速度是很快的。所以在实际使用时，用户自定义的消费逻辑中不应该出现耗时的操作（考虑异步化），否则将可能导致下游的消费者陷入长时间的自旋中浪费大量的CPU资源

MyThreadHints.onSpinWait分析

在waitFor方法中自旋并不是简单的空循环，而是调用了MyThreadHints.onSpinWait方法。

/**  * 启发性的查询是否存在Thread.onSpinWait方法，如果有则可以调用，如果没有则执行空逻辑  *  * 兼容老版本无该方法的jdk（Thread.onSpinWait是jdk9开始引入的）  * */ public class MyThreadHints {      private static final MethodHandle ON_SPIN_WAIT_METHOD_HANDLE;      static {         final MethodHandles.Lookup lookup = MethodHandles.lookup();          MethodHandle methodHandle = null;         try {             methodHandle = lookup.findStatic(Thread.class, "onSpinWait", methodType(void.class));         } catch (final Exception ignore) {             // jdk9才引入的Thread.onSpinWait, 低版本没找到该方法直接忽略异常即可         }          ON_SPIN_WAIT_METHOD_HANDLE = methodHandle;     }      public static void onSpinWait() {         // Call java.lang.Thread.onSpinWait() on Java SE versions that support it. Do nothing otherwise.         // This should optimize away to either nothing or to an inlining of java.lang.Thread.onSpinWait()         if (null != ON_SPIN_WAIT_METHOD_HANDLE) {             try {                 // 如果是高版本jdk找到了Thread.onSpinWait方法，则进行调用, 插入特殊指令优化CPU自旋性能（例如x86架构中的pause汇编指令）                 // invokeExact比起反射调用方法要高一些，详细的原因待研究                 ON_SPIN_WAIT_METHOD_HANDLE.invokeExact();             }             catch (final Throwable ignore) {                 // 异常无需考虑             }         }     } }

MyThreadHints内部在static块中通过MethodHandles尝试着获取Thread类的静态方法onSpinWait。这一方法在jdk9中被引入，因此在包括jdk8在内的低版本jdk中，都无法找到该方法，初始化完后ON_SPIN_WAIT_METHOD_HANDLE会是null。
jdk7开始引入的MethodHandles中，由于提前进行了很多的合法性检查，比常见的反射调用方式效率要高一些（但比起反射缺失了一些好用的功能）。所以在追求极致性能的disruptor中，用于自旋中的逻辑自然选择了效率更高的MethodHandles。
jdk9开始引入的Thread.onSpinWait中，一般是执行特殊的汇编指令用于告诉CPU当前方法处于无意义的自旋等待中，让CPU进行一定的优化。具体的实现主要取决于底层具体的硬件平台类型，如x86架构下是pause指令。
关于pause指令优化CPU自旋性能损耗的原理涉及到过多的硬件知识，限于个人水平就不再展开了。

MyDisruptor v2版本demo解析

v2版本支持了多生产者和消费者组间消费依赖的功能，下面通过一个稍显复杂的demo来展示如何使用这些功能。

public class MyRingBufferV2Demo {      /**      * 树形依赖关系      * A，B->C->E      *    ->D->F,G      * */     public static void main(String[] args) throws InterruptedException {         // 环形队列容量为16（2的4次方）         int ringBufferSize = 16;          // 创建环形队列         MyRingBuffer<OrderEventModel> myRingBuffer = MyRingBuffer.createSingleProducer(                 new OrderEventProducer(), ringBufferSize, new MyBlockingWaitStrategy());          // 获得ringBuffer的序列屏障（最上游的序列屏障内只维护生产者的序列）         MySequenceBarrier mySequenceBarrier = myRingBuffer.newBarrier();          // ================================== 基于生产者序列屏障，创建消费者A         MyBatchEventProcessor<OrderEventModel> eventProcessorA =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerA"), mySequenceBarrier);         MySequence consumeSequenceA = eventProcessorA.getCurrentConsumeSequence();         // RingBuffer监听消费者A的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceA);          // ================================== 基于生产者序列屏障，创建消费者B         MyBatchEventProcessor<OrderEventModel> eventProcessorB =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerB"), mySequenceBarrier);         MySequence consumeSequenceB = eventProcessorB.getCurrentConsumeSequence();         // RingBuffer监听消费者B的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceB);          // ================================== 消费者C依赖上游的消费者A，B，通过消费者A、B的序列号创建序列屏障（构成消费的顺序依赖）         MySequenceBarrier mySequenceBarrierC = myRingBuffer.newBarrier(consumeSequenceA,consumeSequenceB);         // 基于序列屏障，创建消费者C         MyBatchEventProcessor<OrderEventModel> eventProcessorC =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerC"), mySequenceBarrierC);         MySequence consumeSequenceC = eventProcessorC.getCurrentConsumeSequence();         // RingBuffer监听消费者C的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceC);          // ================================== 消费者E依赖上游的消费者C，通过消费者C的序列号创建序列屏障（构成消费的顺序依赖）         MySequenceBarrier mySequenceBarrierE = myRingBuffer.newBarrier(consumeSequenceC);         // 基于序列屏障，创建消费者E         MyBatchEventProcessor<OrderEventModel> eventProcessorE =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerE"), mySequenceBarrierE);         MySequence consumeSequenceE = eventProcessorE.getCurrentConsumeSequence();         // RingBuffer监听消费者E的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceE);          // ================================== 消费者D依赖上游的消费者A,B，通过消费者A、B的序列号创建序列屏障（构成消费的顺序依赖）         MySequenceBarrier mySequenceBarrierD = myRingBuffer.newBarrier(consumeSequenceA,consumeSequenceB);         // 基于序列屏障，创建消费者D         MyBatchEventProcessor<OrderEventModel> eventProcessorD =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerD"), mySequenceBarrierD);         MySequence consumeSequenceD = eventProcessorD.getCurrentConsumeSequence();         // RingBuffer监听消费者D的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceD);          // ================================== 消费者F依赖上游的消费者D，通过消费者D的序列号创建序列屏障（构成消费的顺序依赖）         MySequenceBarrier mySequenceBarrierF = myRingBuffer.newBarrier(consumeSequenceD);         // 基于序列屏障，创建消费者F         MyBatchEventProcessor<OrderEventModel> eventProcessorF =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerF"), mySequenceBarrierF);         MySequence consumeSequenceF = eventProcessorF.getCurrentConsumeSequence();         // RingBuffer监听消费者F的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceF);          // ================================== 消费者G依赖上游的消费者D，通过消费者D的序列号创建序列屏障（构成消费的顺序依赖）         MySequenceBarrier mySequenceBarrierG = myRingBuffer.newBarrier(consumeSequenceD);         // 基于序列屏障，创建消费者G         MyBatchEventProcessor<OrderEventModel> eventProcessorG =                 new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerG"), mySequenceBarrierG);         MySequence consumeSequenceG = eventProcessorG.getCurrentConsumeSequence();         // RingBuffer监听消费者G的序列         myRingBuffer.addGatingConsumerSequenceList(consumeSequenceG);          // 启动消费者线程         new Thread(eventProcessorA).start();         new Thread(eventProcessorB).start();         new Thread(eventProcessorC).start();         new Thread(eventProcessorD).start();         new Thread(eventProcessorE).start();         new Thread(eventProcessorF).start();         new Thread(eventProcessorG).start();          // 生产者发布100个事件         for(int i=0; i<100; i++) {             long nextIndex = myRingBuffer.next();             OrderEventModel orderEvent = myRingBuffer.get(nextIndex);             orderEvent.setMessage("message-"+i);             orderEvent.setPrice(i * 10);             System.out.println("生产者发布事件：" + orderEvent);             myRingBuffer.publish(nextIndex);         }          // 简单阻塞下，避免还未消费完主线程退出         Thread.sleep(5000L);     } }

最上游的的消费者在创建时传入的序列屏障内只维护了生产者序列（myRingBuffer.newBarrier()）。
存在上游依赖的消费者在创建时，其传入的序列屏障内需要维护直接依赖的上游消费者的序列号集合（即dependentSequencesList）用于控制消费速度不超过上游消费者。
每个新创建的消费者都需要通过ringBuffer.addGatingConsumerSequenceList接口将自己的序列号维护到生产者的gatingConsumerSequenceList中，令生产者生产时不会越过最慢的消费者一圈。