disruptor小结--消费者

disruptor主要有以下两种消费模式:

两种消费模式的共同点：消费线程的总数量是一样的，都等于传入的hander的数量，也就是入参多少handler处理器就有多少个线程被创建出来，代码参见：

    //重复消费的处理器代码：private void processEvents(){T event = null;long nextSequence = sequence.get() + 1L;while (true){try{final long availableSequence = sequenceBarrier.waitFor(nextSequence);if (batchStartAware != null){batchStartAware.onBatchStart(availableSequence - nextSequence + 1);}while (nextSequence <= availableSequence){event = dataProvider.get(nextSequence);eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);nextSequence++;}sequence.set(availableSequence);}catch (final TimeoutException e){notifyTimeout(sequence.get());}catch (final AlertException ex){if (running.get() != RUNNING){break;}}catch (final Throwable ex){exceptionHandler.handleEventException(ex, nextSequence, event);sequence.set(nextSequence);nextSequence++;}}}//不重复消费的代码public void run(){if (!running.compareAndSet(false, true)){throw new IllegalStateException("Thread is already running");}sequenceBarrier.clearAlert();notifyStart();boolean processedSequence = true;long cachedAvailableSequence = Long.MIN_VALUE;long nextSequence = sequence.get();T event = null;while (true){try{// if previous sequence was processed - fetch the next sequence and set// that we have successfully processed the previous sequence// typically, this will be true// this prevents the sequence getting too far forward if an exception// is thrown from the WorkHandlerif (processedSequence){processedSequence = false;do{nextSequence = workSequence.get() + 1L;sequence.set(nextSequence - 1L);}while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));}if (cachedAvailableSequence >= nextSequence){event = ringBuffer.get(nextSequence);workHandler.onEvent(event);processedSequence = true;}else{cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);}}catch (final TimeoutException e){notifyTimeout(sequence.get());}catch (final AlertException ex){if (!running.get()){break;}}catch (final Throwable ex){// handle, mark as processed, unless the exception handler threw an exceptionexceptionHandler.handleEventException(ex, nextSequence, event);processedSequence = true;}}notifyShutdown();running.set(false);}

不同点：
1.handler处理器的实现类不一样：多线程重复消费的处理器实现的是EventHandler接口，多线程不重复消费的处理器实现的是WorkHandler接口
2.多线程重复消费的handler实现的关键在于每个handler(BatchEventProcessor)都维护一个sequence，互不干扰，重复消费
多线程不重复消费的handler实现的关键在于所有的handler(workerProcessor)都同享同一个workSequence，所以不会重复消费.

disruptor小结--消费者相关推荐

disruptor小结--生产者代码
disruptor的生产者有两种模式: 分别是单生产者模式和多生产者模式 1.单生产者模式: public long next(int n) { if (n < 1) { throw new I ...
disruptor小结1--优势
disruptor是一个内存队列,目的是为了线程间队列通信使用的,不同于kafka等分布式队列,区别于ArrayBlockQueue等队列的性能优势在于以下几点: a.环形队列(长度是2的n次幂) b ...
disruptor消费者模型
Disruptor的消费者实现是WorkerProcessor类,都实现了EventProcessor接口,这个接口继承了Runnable接口,因此,每个消费者实则就是一条线程,具体的业务逻辑都实现自 ...
生产者消费者模式－java原生、Disruptor实现方案
生产者消费者模式介绍生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题.生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻 ...
JAVA并发编程之 LMAX Disruptor使用实例（高效解决生产者与消费者问题）
什么是Disruptor? Disruptor是一个开源的JAVA框架,它被设计用于在生产者-消费者(producer-consumer problem,简称PCP)问题上获得尽量高的吞吐量(TPS) ...
Disruptor框架中生产者、消费者的各种复杂依赖场景下的使用总结-我见过最好的Disruptor
更多高并发知识请访问 www.itkc8.com 非常感谢 https://www.cnblogs.com/pku-liuqiang/p/8544700.html Disruptor是一个优秀的并发框 ...
Disruptor并发框架--学习笔记
Disruptor并发框架简介 Martin Fowler在自己网站上写了一篇LMAX架构的文章,在文章中他介绍了LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易.这个系统是建立在J ...
Java 并发框架Disruptor（七）
Disruptor VS BlockingQueue的压测对比: import java.util.concurrent.ArrayBlockingQueue;public class ArrayBl ...
disruptor模拟高速处理大规模订单类业务场景
什么是Disruptor? Disruptor是一个开源的JAVA框架,它被设计用于在生产者-消费者(producer-consumer problem,简称PCP)问题上获得尽量高的吞吐量(TPS) ...

disruptor小结--消费者

disruptor小结--消费者相关推荐

最新文章

热门文章