本篇文章为你整理了Disruptor生产和消费模式详解及高级应用（并行模式）（）的详细内容，包含有 Disruptor生产和消费模式详解及高级应用（并行模式），希望能帮助你了解 Disruptor生产和消费模式详解及高级应用（并行模式）。

　　小伙伴们大家好，昨天的文章，带着大家扒开了Disruptor华丽的外衣，最重要的是我们知道了Disruptor高性能的原因几个重要的原因，

　　引入环形的数组结构：数组元素不会被回收，避免频繁的GC，

　　无锁的设计：采用CAS无锁方式，保证线程的安全性

　　属性填充：通过添加额外的无用信息，避免伪共享问题

　　元素位置的定位：采用跟一致性哈希一样的方式，一个索引，进行自增

　　这篇文章就在上篇文章的基础上来点实战应用。研究下Disruptor的生产和消费模式，以及高级应用，至此关于Disruptor的系列的文章，也就到此结束了，我已经尽力了，如果还有什么没能满足大家需求的，以及关于文章的内容大家有任何其他的看法的，也欢迎在评论区留言，毕竟本人才疏学浅，期待各位大佬能给小弟指点一二。

　　前两篇文章在这里哦：

　　生产和消费模式

　　根据上面的环形结构，我们来具体分析一下Disruptor的工作原理。

　　​ Disruptor 不像传统的队列，分为一个队头指针和一个队尾指针，而是只有一个角标（上面的seq），那么这个是如何保证生产的消息不会覆盖没有消费掉的消息呢。

　　​ 在Disruptor中生产者分为单生产者和多生产者，而消费者并没有区分。

　　​ 单生产者情况下，就是普通的生产者向RingBuffer中放置数据，消费者获取最大可消费的位置，并进行消费。而多生产者时候，又多出了一个跟RingBuffer同样大小的Buffer，称为AvailableBuffer。

　　​ 在多生产者中，每个生产者首先通过CAS竞争获取可以写的空间，然后再进行慢慢往里放数据，如果正好这个时候消费者要消费数据，那么每个消费者都需要获取最大可消费的下标，这个下标是在AvailableBuffer进行获取得到的最长连续的序列下标。

　　5.1 单生产者生产数据

　　生产者单线程写数据的流程比较简单：

　　
申请写入m个元素；

　　若是有m个元素可以入，则返回最大的序列号。这儿主要判断是否会覆盖未读的元素；

　　若是返回的正确，则生产者开始写入元素。

　　5.2 多生产者生产数据

　　​ 多个生产者的情况下，会遇到“如何防止多个线程重复写同一个元素”的问题。Disruptor的解决方法是，每个线程获取不同的一段数组空间进行操作。这个通过CAS很容易达到。只需要在分配元素的时候，通过CAS判断一下这段空间是否已经分配出去即可。

　　​ 但是会遇到一个新问题：如何防止读取的时候，读到还未写的元素。Disruptor在多个生产者的情况下，引入了一个与Ring Buffer大小相同的buffer：available Buffer。当某个位置写入成功的时候，便把availble Buffer相应的位置置位，标记为写入成功。读取的时候，会遍历available Buffer，来判断元素是否已经就绪。

　　5.3.1 生产流程

　　申请写入m个元素；

　　若是有m个元素可以写入，则返回最大的序列号。每个生产者会被分配一段独享的空间；

　　生产者写入元素，写入元素的同时设置available Buffer里面相应的位置，以标记自己哪些位置是已经写入成功的。

　　
如下图所示，Writer1和Writer2两个线程写入数组，都申请可写的数组空间。Writer1被分配了下标3到下表5的空间，Writer2被分配了下标6到下标9的空间。

　　Writer1写入下标3位置的元素，同时把available Buffer相应位置置位，标记已经写入成功，往后移一位，开始写下标4位置的元素。Writer2同样的方式。最终都写入完成。

　　5.3.2 CAS检测空间占用

　　防止不同生产者对同一段空间写入的代码，如下所示：

　　​ 通过do/while循环的条件cursor.compareAndSet(current, next)，来判断每次申请的空间是否已经被其他生产者占据。假如已经被占据，该函数会返回失败，While循环重新执行，申请写入空间。

public long tryNext(int n) throws InsufficientCapacityException

　　 if (n 1)

　　 throw new IllegalArgumentException("n must be 0");

　　 long current;

　　 long next;

　　 current = cursor.get();

　　 next = current + n;

　　 if (!hasAvailableCapacity(gatingSequences, n, current))

　　 throw InsufficientCapacityException.INSTANCE;

　　 while (!cursor.compareAndSet(current, next));

　　 return next;

　　5.4 多生产者消费数据

　　绿色代表已经写OK的数据

　　​ 假设三个生产者在写中，还没有置位AvailableBuffer，那么消费者可获取的消费下标只能获取到6，然后等生产者都写OK后，通知到消费者，消费者继续重复上面的步骤。

　　5.4.1 消费流程

　　申请读取到序号n；

　　若writer cursor = n，这时仍然无法确定连续可读的最大下标。从reader cursor开始读取available Buffer，一直查到第一个不可用的元素，然后返回最大连续可读元素的位置；

　　消费者读取元素

　　
如下图所示，读线程读到下标为2的元素，三个线程Writer1/Writer2/Writer3正在向RingBuffer相应位置写数据，写线程被分配到的最大元素下标是11。

　　读线程申请读取到下标从3到11的元素，判断writer cursor =11。然后开始读取availableBuffer，从3开始，往后读取，发现下标为7的元素没有生产成功，于是WaitFor(11)返回6。

　　然后，消费者读取下标从3到6共计4个元素。

　　6. 高级使用

　　6.1 并行模式

　　6.1.1 单一写者模式

　　​ 在并发系统中提高性能最好的方式之一就是单一写者原则，对Disruptor也是适用的。如果在你的代码中仅仅有一个事件生产者，那么可以设置为单一生产者模式来提高系统的性能。

public class singleProductorLongEventMain { 

　　 public static void main(String[] args) throws Exception {

　　 //.....// Construct the Disruptor with a SingleProducerSequencer

　　 Disruptor LongEvent disruptor = new Disruptor(factory,

　　 bufferSize,

　　 ProducerType.SINGLE, // 单一写者模式,

　　 executor);//.....

　　6.1.2 串行消费

　　比如：现在触发一个注册Event，需要有一个Handler来存储信息，一个Hanlder来发邮件等等。

/**

　　 * 串行依次执行

　　 * br/

　　 * p -- c11 -- c21

　　 * @param disruptor

　　 public static void serial(Disruptor LongEvent disruptor){

　　 disruptor.handleEventsWith(new C11EventHandler()).then(new C21EventHandler());

　　 disruptor.start();

　　6.1.3 菱形方式执行

 public static void diamond(Disruptor LongEvent disruptor){

　　 disruptor.handleEventsWith(new C11EventHandler(),new C12EventHandler()).then(new C21EventHandler());

　　 disruptor.start();

　　6.1.4 链式并行计算

 public static void chain(Disruptor LongEvent disruptor){

　　 disruptor.handleEventsWith(new C11EventHandler()).then(new C12EventHandler());

　　 disruptor.handleEventsWith(new C21EventHandler()).then(new C22EventHandler());

　　 disruptor.start();

　　6.1.5 相互隔离模式

 public static void parallelWithPool(Disruptor LongEvent disruptor){

　　 disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler());

　　 disruptor.handleEventsWithWorkerPool(new C21EventHandler(),new C21EventHandler());

　　 disruptor.start();

　　6.1.6 航道模式

/**

　　 * 串行依次执行,同时C11，C21分别有2个实例

　　 * br/

　　 * p -- c11 -- c21

　　 * @param disruptor

　　 public static void serialWithPool(Disruptor LongEvent disruptor){

　　 disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler()).then(new C21EventHandler(),new C21EventHandler());

　　 disruptor.start();

　　本文由传智教育博学谷教研团队发布。

　　如果本文对您有帮助，欢迎关注和点赞；如果您有任何建议也可留言评论或私信，您的支持是我坚持创作的动力。

　　转载请注明出处！

　　以上就是Disruptor生产和消费模式详解及高级应用（并行模式）（）的详细内容，想要了解更多 Disruptor生产和消费模式详解及高级应用（并行模式）的内容，请持续关注盛行IT软件开发工作室。

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。