?

單一的生產(chǎn)者，消費者有多個，使用WorkerPool來管理多個消費者；

?

RingBuffer在生產(chǎn)Sequencer中記錄一個cursor，追蹤生產(chǎn)者生產(chǎn)到的最新位置，通過WorkSequence和sequence記錄整個workpool消費的位置和每個WorkProcessor消費到位置，來協(xié)調(diào)生產(chǎn)和消費程序

?

1、定義事件

package com.ljq.disruptor;import java.io.Serializable;/*** 交易事件數(shù)據(jù)* * @author Administrator**/ @SuppressWarnings("serial") public class TradeEvent implements Serializable {private String id; // 訂單IDprivate String name;private double price; // 金額public TradeEvent() {}public TradeEvent(String id) {super();this.id = id;}public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}@Overridepublic String toString() {return "Trade [id=" + id + ", name=" + name + ", price=" + price + "]";}}

?

2、TradeEvent事件消費者

package com.ljq.disruptor;import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler;public class TradeEventHandler implements EventHandler<TradeEvent>, WorkHandler<TradeEvent> {@Overridepublic void onEvent(TradeEvent event, long sequence, boolean endOfBatch) throws Exception {this.onEvent(event);}/*** WorkProcessor多線程排隊領(lǐng)event然后再執(zhí)行，不同線程執(zhí)行不同的event。但是多了個排隊領(lǐng)event的過程，這個是為了減少對生產(chǎn)者隊列查詢的壓力*/@Overridepublic void onEvent(TradeEvent event) throws Exception {// 具體的消費邏輯System.out.println("consumer:" + Thread.currentThread().getName() + " Event: value=" + event);} }

?

3、EventProcessor消費者-生產(chǎn)者啟動類

package com.ljq.disruptor;import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future;import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventFactory; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; import com.lmax.disruptor.YieldingWaitStrategy;public class EventProcessorMain {public static void main(String[] args) throws Exception { long beginTime = System.currentTimeMillis();// 指定 ring buffer字節(jié)大小，必需為2的N次方(能將求模運算轉(zhuǎn)為位運算提高效率 )，否則影響性能int bufferSize = 1024;//固定線程數(shù)int nThreads = 4;EventFactory<TradeEvent> eventFactory = new EventFactory<TradeEvent>() { @Override public TradeEvent newInstance() { return new TradeEvent(UUID.randomUUID().toString());} };//RingBuffer. createSingleProducer創(chuàng)建一個單生產(chǎn)者的RingBuffer//第一個參數(shù)叫EventFactory，從名字上理解就是“事件工廠”，其實它的職責(zé)就是產(chǎn)生數(shù)據(jù)填充RingBuffer的區(qū)塊。 //第二個參數(shù)是RingBuffer的大小，它必須是2的整數(shù)倍，目的是為了將求模運算轉(zhuǎn)為&運算提高效率//第三個參數(shù)是RingBuffer的生產(chǎn)在沒有可用區(qū)塊的時候(可能是消費者太慢了)的等待策略 final RingBuffer<TradeEvent> ringBuffer = RingBuffer.createSingleProducer(eventFactory, bufferSize, new YieldingWaitStrategy()); //SequenceBarrier, 協(xié)調(diào)消費者與生產(chǎn)者, 消費者鏈的先后順序. 阻塞后面的消費者(沒有Event可消費時)SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(); //創(chuàng)建消費者事件處理器， 多線程并發(fā)執(zhí)行，不同線程執(zhí)行不同的event BatchEventProcessor<TradeEvent> transProcessor = new BatchEventProcessor<TradeEvent>(ringBuffer, sequenceBarrier, new TradeEventHandler()); //把消費者的消費進度情況注冊給RingBuffer結(jié)構(gòu)(生產(chǎn)者)，如果只有一個消費者的情況可以省略 ringBuffer.addGatingSequences(transProcessor.getSequence()); //創(chuàng)建一個可重用固定線程數(shù)的線程池，以共享的無界隊列方式來運行這些線程ExecutorService executors = Executors.newFixedThreadPool(nThreads); //把消費者提交到線程池，說明EventProcessor實現(xiàn)了callable接口 executors.submit(transProcessor); // 生產(chǎn)者，這里新建線程不是必要的Future<?> future= executors.submit(new Callable<Void>() { @Override public Void call() throws Exception { long seq; for (int i = 0; i < 100000; i++) {seq = ringBuffer.next();ringBuffer.get(seq).setPrice(i);ringBuffer.publish(seq);} return null; } }); future.get();//等待生產(chǎn)者結(jié)束 Thread.sleep(1000); //等上1秒，等消費都處理完成transProcessor.halt(); //通知事件(或者說消息)處理器 可以結(jié)束了（并不是馬上結(jié)束!!!） executors.shutdown(); System.out.println(String.format("總共耗時%s毫秒", (System.currentTimeMillis() - beginTime)));} }

?

4、WorkerPool消費者-生產(chǎn)者啟動類

package com.ljq.disruptor;import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;import com.lmax.disruptor.EventFactory; import com.lmax.disruptor.IgnoreExceptionHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; import com.lmax.disruptor.WorkerPool;public class WorkPoolMain {public static void main(String[] args) throws InterruptedException {// 指定 ring buffer字節(jié)大小，必需為2的N次方(能將求模運算轉(zhuǎn)為位運算提高效率 )，否則影響性能int bufferSize = 1024;//固定線程數(shù)int nThreads = 4;//RingBuffer. createSingleProducer創(chuàng)建一個單生產(chǎn)者的RingBufferRingBuffer<TradeEvent> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<TradeEvent>() {public TradeEvent newInstance() {return new TradeEvent(UUID.randomUUID().toString());}}, bufferSize);SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();WorkerPool<TradeEvent> workerPool = new WorkerPool<TradeEvent>(ringBuffer, sequenceBarrier,new IgnoreExceptionHandler(), new TradeEventHandler());//創(chuàng)建一個可重用固定線程數(shù)的線程池，以共享的無界隊列方式來運行這些線程ExecutorService executors = Executors.newFixedThreadPool(nThreads);workerPool.start(executors);// 生產(chǎn)10個數(shù)據(jù)for (int i = 0; i < 80000; i++) {long seq = ringBuffer.next();ringBuffer.get(seq).setPrice(i);ringBuffer.publish(seq);}Thread.sleep(1000); //等上1秒，等消費都處理完成workerPool.halt(); //通知事件(或者說消息)處理器 可以結(jié)束了（并不是馬上結(jié)束!!!） executors.shutdown(); } }

?

總結(jié)

以上是生活随笔為你收集整理的EventProcessor与WorkPool用法--可处理多消费者的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。