并发框架Disruptor几个Demo
经过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)); } }
上一篇: 历史上第一个驸马竟然是个外国人?“驸马”一称是怎么来的?
下一篇: JS函数 0316