?

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

?

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

?

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事件消費(fèi)者

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多線(xiàn)程排隊(duì)領(lǐng)event然后再執(zhí)行，不同線(xiàn)程執(zhí)行不同的event。但是多了個(gè)排隊(duì)領(lǐng)event的過(guò)程，這個(gè)是為了減少對(duì)生產(chǎn)者隊(duì)列查詢(xún)的壓力*/@Overridepublic void onEvent(TradeEvent event) throws Exception {// 具體的消費(fèi)邏輯System.out.println("consumer:" + Thread.currentThread().getName() + " Event: value=" + event);} }

?

3、EventProcessor消費(fèi)者-生產(chǎn)者啟動(dòng)類(lèi)

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次方(能將求模運(yùn)算轉(zhuǎn)為位運(yùn)算提高效率 )，否則影響性能int bufferSize = 1024;//固定線(xiàn)程數(shù)int nThreads = 4;EventFactory<TradeEvent> eventFactory = new EventFactory<TradeEvent>() { @Override public TradeEvent newInstance() { return new TradeEvent(UUID.randomUUID().toString());} };//RingBuffer. createSingleProducer創(chuàng)建一個(gè)單生產(chǎn)者的RingBuffer//第一個(gè)參數(shù)叫EventFactory，從名字上理解就是“事件工廠”，其實(shí)它的職責(zé)就是產(chǎn)生數(shù)據(jù)填充RingBuffer的區(qū)塊。 //第二個(gè)參數(shù)是RingBuffer的大小，它必須是2的整數(shù)倍，目的是為了將求模運(yùn)算轉(zhuǎn)為&運(yùn)算提高效率//第三個(gè)參數(shù)是RingBuffer的生產(chǎn)在沒(méi)有可用區(qū)塊的時(shí)候(可能是消費(fèi)者太慢了)的等待策略 final RingBuffer<TradeEvent> ringBuffer = RingBuffer.createSingleProducer(eventFactory, bufferSize, new YieldingWaitStrategy()); //SequenceBarrier, 協(xié)調(diào)消費(fèi)者與生產(chǎn)者, 消費(fèi)者鏈的先后順序. 阻塞后面的消費(fèi)者(沒(méi)有Event可消費(fèi)時(shí))SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(); //創(chuàng)建消費(fèi)者事件處理器， 多線(xiàn)程并發(fā)執(zhí)行，不同線(xiàn)程執(zhí)行不同的event BatchEventProcessor<TradeEvent> transProcessor = new BatchEventProcessor<TradeEvent>(ringBuffer, sequenceBarrier, new TradeEventHandler()); //把消費(fèi)者的消費(fèi)進(jìn)度情況注冊(cè)給RingBuffer結(jié)構(gòu)(生產(chǎn)者)，如果只有一個(gè)消費(fèi)者的情況可以省略 ringBuffer.addGatingSequences(transProcessor.getSequence()); //創(chuàng)建一個(gè)可重用固定線(xiàn)程數(shù)的線(xiàn)程池，以共享的無(wú)界隊(duì)列方式來(lái)運(yùn)行這些線(xiàn)程ExecutorService executors = Executors.newFixedThreadPool(nThreads); //把消費(fèi)者提交到線(xiàn)程池，說(shuō)明EventProcessor實(shí)現(xiàn)了callable接口 executors.submit(transProcessor); // 生產(chǎn)者，這里新建線(xià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秒，等消費(fèi)都處理完成transProcessor.halt(); //通知事件(或者說(shuō)消息)處理器 可以結(jié)束了（并不是馬上結(jié)束!!!） executors.shutdown(); System.out.println(String.format("總共耗時(shí)%s毫秒", (System.currentTimeMillis() - beginTime)));} }

?

4、WorkerPool消費(fèi)者-生產(chǎn)者啟動(dòng)類(lèi)

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次方(能將求模運(yùn)算轉(zhuǎn)為位運(yùn)算提高效率 )，否則影響性能int bufferSize = 1024;//固定線(xiàn)程數(shù)int nThreads = 4;//RingBuffer. createSingleProducer創(chuàng)建一個(gè)單生產(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)建一個(gè)可重用固定線(xiàn)程數(shù)的線(xiàn)程池，以共享的無(wú)界隊(duì)列方式來(lái)運(yùn)行這些線(xiàn)程ExecutorService executors = Executors.newFixedThreadPool(nThreads);workerPool.start(executors);// 生產(chǎn)10個(gè)數(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秒，等消費(fèi)都處理完成workerPool.halt(); //通知事件(或者說(shuō)消息)處理器 可以結(jié)束了（并不是馬上結(jié)束!!!） executors.shutdown(); } }

?

總結(jié)

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

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