经过2个月的疯狂加班后终于迎来了五一节前的几天清闲日子，在这闲得蛋疼的日子里，本屌丝无意中从ifeve上看到“Disruptor”并发框架，它号称"能够在一个线程里每秒处理6百万订单" 当时就被它吸引住了，不管是不是在吹流逼吧，先研究研究在说....
扫盲：
要想了解Disruptor框架必需多花点时间研究下它的工作原理，不然代码是没法撸的！！！
关于Disruptor的详细资料及原理请细看!!! http://ifeve.com/disruptor/
Disruptor版本： 3.2.1
名词解释
消费者==事件处理器
一、现在你必须明白以下问题：
1、你必须明白Ringbuffer是什么，它的数据结构是怎么样的，有什么约定，为什么高效，它的职责是什么。
2、ConsumerBarrier （ifeve网上的译文版本比较早，这个类在2.0.0之后就一直被改名，3.2.1的版本中它是SequenceBarrier）它的职责是什么。
Disruptor框架在2.0版本之后不再采用生产者、消费者模型来编写API，而是使用事件模型，其实只是接口设计、类名和概念上的变化，内部原理其实还是一样的。
DEMO 一、使用原生API创建一个简单的生产者和消费者
//DEMO中使用的 消息全假定是一条交易
public class TradeTransaction {
private String id;//交易ID
private double price;//交易金额
public TradeTransaction() {
}
public TradeTransaction(String id, double price) {
super();
this.id = id;
this.price = price;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public double getPrice() {
return price;
}
public void setPrice(double price) {
this.price = price;
}
}
public class TradeTransactionInDBHandler implements EventHandler<TradeTransaction>,WorkHandler<TradeTransaction> {
@Override
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
this.onEvent(event);
}
@Override
public void onEvent(TradeTransaction event) throws Exception {
//这里做具体的消费逻辑
event.setId(UUID.randomUUID().toString());//简单生成下ID
System.out.println(event.getId());
}
}
public class Demo1 {
public static void main(String[] args) throws InterruptedException, ExecutionException {
int BUFFER_SIZE=1024;
int THREAD_NUMBERS=4;
/*
* createSingleProducer创建一个单生产者的RingBuffer，
* 第一个参数叫EventFactory，从名字上理解就是“事件工厂”，其实它的职责就是产生数据填充RingBuffer的区块。
* 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
* 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
*/
final RingBuffer<TradeTransaction> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<TradeTransaction>() {
@Override
public TradeTransaction newInstance() {
return new TradeTransaction();
}
}, BUFFER_SIZE,new YieldingWaitStrategy());
//创建线程池
ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
//创建SequenceBarrier
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
//创建消息处理器
BatchEventProcessor<TradeTransaction> transProcessor = new BatchEventProcessor<TradeTransaction>(
ringBuffer, sequenceBarrier, new TradeTransactionInDBHandler());
//这一部的目的是让RingBuffer根据消费者的状态 如果只有一个消费者的情况可以省略
ringBuffer.addGatingSequences(transProcessor.getSequence());
//把消息处理器提交到线程池
executors.submit(transProcessor);
//如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类
Future<?> future=executors.submit(new Callable<Void>() {
@Override
public Void call() throws Exception {
long seq;
for(int i=0;i<1000;i++){
seq=ringBuffer.next();//占个坑 --ringBuffer一个可用区块
ringBuffer.get(seq).setPrice(Math.random()*9999);//给这个区块放入 数据 如果此处不理解，想想RingBuffer的结构图
ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见
}
return null;
}
});
future.get();//等待生产者结束
Thread.sleep(1000);//等上1秒，等消费都处理完成
transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
executors.shutdown();//终止线程
}
}
DEMO二、使用WorkerPool辅助创建消费者
public class Demo2 {
public static void main(String[] args) throws InterruptedException {
int BUFFER_SIZE=1024;
int THREAD_NUMBERS=4;
EventFactory<TradeTransaction> eventFactory=new EventFactory<TradeTransaction>() {
public TradeTransaction newInstance() {
return new TradeTransaction();
}
};
RingBuffer<TradeTransaction> ringBuffer=RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);
WorkHandler<TradeTransaction> workHandlers=new TradeTransactionInDBHandler();
/*
* 这个类代码很简单的，亲自己看哈！~
*/
WorkerPool<TradeTransaction> workerPool=new WorkerPool<TradeTransaction>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), workHandlers);
workerPool.start(executor);
//下面这个生产8个数据，图简单就写到主线程算了
for(int i=0;i<8;i++){
long seq=ringBuffer.next();
ringBuffer.get(seq).setPrice(Math.random()*9999);
ringBuffer.publish(seq);
}
Thread.sleep(1000);
workerPool.halt();
executor.shutdown();
}
}
DEMO三、demo3写个流弊点的像下图这样。这次用Disruptor来完成整个构建工作.
从中图可以看出需求是介样子的：生产者生产数据经过C1,C2处理完成后再到C3。
假设如下场景：
1、交易网关收到交易(P1)把交易数据发到RingBuffer中，
2、负责处理增值业务的消费者C1和负责数据存储的消费者C2负责处理交易
3、负责发送JMS消息的消费者C3在C1和C2处理完成后再进行处理。
让代码说话：
public class TradeTransactionJMSNotifyHandler implements EventHandler<TradeTransaction> {
@Override
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
//do send jms message
}
}
public class TradeTransactionPublisher implements Runnable{
Disruptor<TradeTransaction> disruptor;
private CountDownLatch latch;
private static int LOOP=10000000;//模拟一千万次交易的发生
public TradeTransactionPublisher(CountDownLatch latch,Disruptor<TradeTransaction> disruptor) {
this.disruptor=disruptor;
this.latch=latch;
}
@Override
public void run() {
TradeTransactionEventTranslator tradeTransloator=new TradeTransactionEventTranslator();
for(int i=0;i<LOOP;i++){
disruptor.publishEvent(tradeTransloator);
}
latch.countDown();
}
}
class TradeTransactionEventTranslator implements EventTranslator<TradeTransaction>{
private Random random=new Random();
@Override
public void translateTo(TradeTransaction event, long sequence) {
this.generateTradeTransaction(event);
}
private TradeTransaction generateTradeTransaction(TradeTransaction trade){
trade.setPrice(random.nextDouble()*9999);
return trade;
}
}
public class TradeTransactionVasConsumer implements EventHandler<TradeTransaction> {
@Override
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
//do something....
}
}
public class Demo3 {
public static void main(String[] args) throws InterruptedException {
long beginTime=System.currentTimeMillis();
int bufferSize=1024;
ExecutorService executor=Executors.newFixedThreadPool(4);
//这个构造函数参数，相信你在了解上面2个demo之后就看下就明白了，不解释了~
Disruptor<TradeTransaction> disruptor=new Disruptor<TradeTransaction>(new EventFactory<TradeTransaction>() {
@Override
public TradeTransaction newInstance() {
return new TradeTransaction();
}
}, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
//使用disruptor创建消费者组C1,C2
EventHandlerGroup<TradeTransaction> handlerGroup=disruptor.handleEventsWith(new TradeTransactionVasConsumer(),new TradeTransactionInDBHandler());
TradeTransactionJMSNotifyHandler jmsConsumer=new TradeTransactionJMSNotifyHandler();
//声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
handlerGroup.then(jmsConsumer);
disruptor.start();//启动
CountDownLatch latch=new CountDownLatch(1);
//生产者准备
executor.submit(new TradeTransactionPublisher(latch, disruptor));
latch.await();//等待生产者完事.
disruptor.shutdown();
executor.shutdown();
System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
}
}
这个图片的流程，应该咋写代码呢，求楼主赐教，刚接触Disruptor。
相关推荐
Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip
### Disruptor并发框架知识点详解 #### 一、Disruptor简介及原理 **Disruptor** 是一款高性能、低延迟的并发框架，它通过无锁设计实现了高效的队列操作，从而大大提升了多线程环境下的性能表现。该框架于2011年...
Disruptor，由LMAX公司开发并开源，是一个高性能、低延迟的并发框架，尤其适用于需要大量数据交换的系统。本文将详细介绍Disruptor的核心原理、设计模式及其在实际中的应用。 Disruptor的诞生源于LMAX对金融交易...
Java并发框架Disruptor，里面采取环形缓存结构，速度更快,适用于生产者消费者模式
Disruptor 是一个 Java 的并发编程框架，大大的简化了并发程序开发的难度，在性能上也比 Java 本身提供的一些并发包要好。 标签：Disruptor
"DisruptorDemo.zip"的实例代码为我们提供了学习和理解Disruptor的一个良好起点，通过实际操作，我们可以更直观地感受Disruptor的强大性能。在实际项目中，尤其是对于需要处理大量并发请求的系统，Disruptor是一个...
本项目是一个基于Spring Boot和LMAX Disruptor框架的高性能并发框架，旨在提供高效的事件处理和消息传递机制。项目涵盖了并发编程的核心概念、无锁并行计算框架的使用、高级特性、底层源码分析以及与Netty的整合实战...
通过这个"Disruptor demo"，你可以学习到如何在实际项目中应用Disruptor来优化并发性能，以及如何配置和调整各种参数以适应不同的系统需求。这个示例将帮助你理解Disruptor如何简化并发编程，提高程序运行效率，尤其...
Disruptor，由LMAX公司开源的一款并发框架，为处理高并发场景提供了一种新颖且高效的解决方案。它通过消除锁和线程间通信的开销，实现了微秒级的延迟和极高的吞吐量，尤其适用于金融交易、实时分析等对性能有苛刻...
SpringBoot整合Disruptor并发编程框架是针对高并发场景下性能优化的一种技术实践。Disruptor是由LMAX公司开发的一款高性能、低延迟的并发工具，它通过消除线程间的锁竞争，大大提升了多线程环境下的处理速度。...
Disruptor它是一个开源的并发框架能够在无锁的情况下实现网络的Queue并发操作。同时，Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者...
《Disruptor并发框架详解》 在高性能计算和并发编程领域，Disruptor是一个不可或缺的名字。这个由LMAX公司开发的并发框架，以其卓越的性能和低延迟特性，被广泛应用于金融交易、实时数据处理等领域。本文将深入探讨...
《Disruptor 框架详解与应用实例》 ...总的来说，这个压缩包提供了一个了解和学习 Disruptor 的良好起点，通过阅读源码、示例代码和 API 文档，我们可以深入了解这个框架如何帮助我们构建高并发、低延迟的应用系统。
在官方原有基础上为降低cpu占用修改版本，性能方面没有任何影响
Disruptor是一个高性能的无锁并发框架，最初由LMAX开发。与传统的基于锁的并发模型相比，Disruptor通过创新的设计大幅提升了数据处理的效率和吞吐量。其核心特性包括环形数组结构、元素位置定位、无锁设计、和多种...
综上所述，LMAX Disruptor是一个强大的并发工具，通过其独特的设计和优化，为高并发场景提供了高效的解决方案。了解和掌握Disruptor，能够帮助开发者构建更高效、更稳定的系统。通过提供的API、源码和示例，开发者...
这个系统是建立在JVM平台上，其核心是一个业务逻辑处理 器，它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中，使用事件源驱动方 式。业务逻辑处理器的核心是Disruptor。
Disruptor是由LMAX公司开发的一种高性能的并发编程框架，主要应用于金融交易系统。它通过优化数据共享方式，显著提高了多线程环境下的处理速度。在"Disruptor 多个消费者"的场景中，我们可以深入理解Disruptor如何...
首先，我们来看看`DisruptorDemo`。Disruptor是LMAX公司开源的一个高性能事件处理框架，它通过避免传统锁机制来实现线程间的通信，极大地降低了数据同步的开销。Disruptor使用环形缓冲区（Ring Buffer）作为数据结构...
简单讲解disruptor并附上demo