disruptor笔记之二：Disruptor类分析

桃扇骨 2022-10-16 05:53 198阅读 0赞

### 欢迎访问我的GitHub ###

> 这里分类和汇总了欣宸的全部原创(含配套源码)：[https://github.com/zq2599/blog\_demos][https_github.com_zq2599_blog_demos]

### 《disruptor笔记》系列链接 ###

1.  [快速入门][Link 1]
2.  [Disruptor类分析][Disruptor]
3.  [环形队列的基础操作(不用Disruptor类)][Disruptor 1]
4.  [事件消费知识点小结][Link 2]
5.  [事件消费实战][Link 3]
6.  [常见场景][Link 4]
7.  [等待策略][Link 5]
8.  [知识点补充(终篇)][Link 6]

### 本篇概览 ###

*  通过[前文][Link 7]的实战，咱们对Disruptor有了初步认识，借助com.lmax.disruptor.dsl.Disruptor类可以轻松完成以下操作：

1.  环形队列初始化
2.  指定事件消费者
3.  启动消费者线程

*  接下来要面对两个问题：

1.  深入了解Disruptor类是如何完成上述操作的；
2.  对Disruptor类有了足够了解时，尝试不用Disruptor，自己动手操作环形队列，实现消息的生产和消费，这样做的目的是加深对Disruptor内部的认识，做到知其所以然；

*  接下来咱们先解决第一个问题吧，结合Disruptor对象的源码来看看上述三个操作到底做了什么；

### 环形队列初始化 ###

*  环形队列初始化发生在实例化Disruptor对象的时候，即Disruptor的构造方法：

public Disruptor(final EventFactory<T> eventFactory, final int ringBufferSize, final ThreadFactory threadFactory)
        { 
            this(RingBuffer.createMultiProducer(eventFactory, ringBufferSize), new BasicExecutor(threadFactory));
        }

*  RingBuffer.createMultiProducer方法内部实例化了RingBuffer，如下图红框：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]
 *  记下第一个重要知识点：创建RingBuffer对象；

### 指定事件消费者 ###

*  在前文中，下面这行代码指定了事件由StringEventHandler消费：

disruptor.handleEventsWith(new StringEventHandler(eventCountPrinter));

*  查看handleEventsWith方法的内部：

public final EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers)
    { 
    	return createEventProcessors(new Sequence[0], handlers);
    }

*  展开createEventProcessors方法，如下图，请重点关注创建SequenceBarrier和BatchEventProcessor等操作：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 1]
 *  展开上图红框四中的updateGatingSequencesForNextInChain方法，如下图，红框中的ringBuffer.addGatingSequences需要重点关注：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 2]
 *  小结一下，disruptor.handleEventsWith方法涉及到四个重要知识点：

1.  创建SequenceBarrier对象，用于接收ringBuffer中的可消费事件
2.  创建BatchEventProcessor，负责消费事件
3.  绑定BatchEventProcessor对象的异常处理类
4.  调用ringBuffer.addGatingSequences，将消费者的Sequence传给ringBuffer

### 启动消费者线程 ###

*  前文已通过日志确定了消费事件的逻辑是在一个独立的线程中执行的，启动消费者线程的代码如下：

disruptor.start();

*  展开start方法，如下可见，关键代码是consumerInfo.start(executor)：

public RingBuffer<T> start()
        { 
            checkOnlyStartedOnce();
            for (final ConsumerInfo consumerInfo : consumerRepository)
            { 
                consumerInfo.start(executor);
            }
    
            return ringBuffer;
        }

*  ConsumerInfo是接口，对应的实现类有EventProcessorInfo和WorkerPoolInfo两种，这里应该是哪种呢？既然来源是consumerRepository，这就要看当初是怎么存入consumerRepository的，前面在分析createEventProcessors方法时，下图红框中的consumerRepository.add被忽略了，现在需要进去看看：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 3]
 *  进去后一目了然，可见ConsumerInfo的实现是EventProcessorInfo：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 4]
 *  所以，回到前面对consumerInfo.start(executor)方法的分析，这里要看的就是EventProcessorInfo的start方法了，如下图，非常简单，就是启动一个线程执行eventprocessor（这个eventprocessor是BatchEventProcessor对象）：  
    ![在这里插入图片描述][20210529093549521.png_pic_center]
 *  小结一下，disruptor.start方法涉及到一个重要知识点：

1.  启动独立线程，用来执行消费事件的业务逻辑；

### 消费事件的逻辑 ###

*  为了理解消息处理逻辑，还要重点关注BatchEventProcessor.processEvents方法，如下图所示，其实也很简单，就是不停的从环形队列取出可用的事件，然后再更新自己的Sequence，相当于标记已经消费到哪里了：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 5]

### 总结 ###

最后总结Disruptor类的重要功能：

1.  创建环形队列（RingBuffer对象）
2.  创建SequenceBarrier对象，用于接收ringBuffer中的可消费事件
3.  创建BatchEventProcessor，负责消费事件
4.  绑定BatchEventProcessor对象的异常处理类
5.  调用ringBuffer.addGatingSequences，将消费者的Sequence传给ringBuffer
6.  启动独立线程，用来执行消费事件的业务逻辑

*  聪明的您一定会发现，本文并没有全面分析Disruptor类的源码，例如after、shutdown等方法都没有提到，确实如此，欣宸在此给您道歉了，本篇的重点是找出那些与基本功能有关代码，为后面的实战提供理论指导（不用Disruptor类实现消息生产消费的实战），因此很多高级功能都跳过了；

### 理解官方流程图 ###

*  此时再看官方流程图，聪明的您应该很快就能理解此图表达的意思：每个消费者都有自己的Sequence，通过此Sequence取得自己在环形队列中消费的位置，再通过SequenceBarrier来等待可用事件的出现，等到事件出现了就用get方法取出具体的事件，给EventHandler来处理：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 6]

### 后续预告 ###

*  此时，咱们对Disruptor类已经有了比较深入的理解，接下来的文章，咱们会尝试不用Disruptor类，仅凭着对RingBuffer对象的操作来实现以下三种功能：

1.  100个事件，单个消费者消费；
2.  100个事件，三个消费者，每个都独自消费这个100个事件；
3.  100个事件，三个消费者共同消费这个100个事件；

### 你不孤单，欣宸原创一路相伴 ###

1.  [Java系列][Java]
2.  [Spring系列][Spring]
3.  [Docker系列][Docker]
4.  [kubernetes系列][kubernetes]
5.  [数据库+中间件系列][Link 8]
6.  [DevOps系列][DevOps]

[https_github.com_zq2599_blog_demos]: https://github.com/zq2599/blog_demos
[Link 1]: https://blog.csdn.net/boling_cavalry/article/details/117185656
[Disruptor]: https://blog.csdn.net/boling_cavalry/article/details/117318462
[Disruptor 1]: https://blog.csdn.net/boling_cavalry/article/details/117386253
[Link 2]: https://blog.csdn.net/boling_cavalry/article/details/117395009
[Link 3]: https://blog.csdn.net/boling_cavalry/article/details/117405835
[Link 4]: https://blog.csdn.net/boling_cavalry/article/details/117575447
[Link 5]: https://blog.csdn.net/boling_cavalry/article/details/117608051
[Link 6]: https://blog.csdn.net/boling_cavalry/article/details/117636483
[Link 7]: https://xinchen.blog.csdn.net/article/details/117185656
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221014/0b7e60c4883c4fdcb47341eb2af21d28.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221014/3595ab164bcb48f0acbfb7c4c4ee7c3e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221014/820f447ed80f465b8b7dbea6962f58bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221014/ccd285da71ce4201a79079efe1c9f482.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221014/b6884c6f7f29431f8bfd04f57ad3f8c8.png
[20210529093549521.png_pic_center]: /images/20221014/592dbef7c3294da48cc0d5b1bda8e9d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221014/1041b88578a24ced9ae216dd8cdd1e3e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221014/2236393676fe41ee8b83c9d46b1f6df1.png
[Java]: https://xinchen.blog.csdn.net/article/details/105068742
[Spring]: https://xinchen.blog.csdn.net/article/details/105086498
[Docker]: https://xinchen.blog.csdn.net/article/details/105086732
[kubernetes]: https://xinchen.blog.csdn.net/article/details/105086794
[Link 8]: https://xinchen.blog.csdn.net/article/details/105086850
[DevOps]: https://xinchen.blog.csdn.net/article/details/105086920