併發框架Disruptor幾個Demo

時間 2019-11-08

標籤併發框架 disruptor 幾個 demo 简体版

原文原文鏈接

掃盲：

要想了解Disruptor框架必需多花點時間研究下它的工做原理，否則代碼是無法擼的！！！數據結構

關於Disruptor的詳細資料及原理請細看!!! http://ifeve.com/disruptor/ 框架

Disruptor版本： 3.2.1

名詞解釋

消費者==事件處理器 dom

1、如今你必須明白如下問題：

一、你必須明白Ringbuffer是什麼，它的數據結構是怎麼樣的，有什麼約定，爲何高效，它的職責是什麼。 ide

二、ConsumerBarrier （ifeve網上的譯文版本比較早，這個類在2.0.0以後就一直被更名，3.2.1的版本中它是SequenceBarrier）它的職責是什麼。函數

Disruptor框架在2.0版本以後再也不採用生產者、消費者模型來編寫API，而是使用事件模型，其實只是接口設計、類名和概念上的變化，內部原理其實仍是同樣的。 this

DEMO 1、使用原生API建立一個簡單的生產者和消費者

Java代碼

//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();//終止線程
}
}

DEMO2、使用WorkerPool輔助建立消費者

Java代碼

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();
}
}

DEMO3、demo3寫個流弊點的像下圖這樣。此次用Disruptor來完成整個構建工做.

從中圖能夠看出需求是介樣子的：生產者生產數據通過C1,C2處理完成後再到C3。 spa

假設以下場景：線程

一、交易網關收到交易(P1)把交易數據發到RingBuffer中，設計

二、負責處理增值業務的消費者C1和負責數據存儲的消費者C2負責處理交易接口

三、負責發送JMS消息的消費者C3在C1和C2處理完成後再進行處理。

讓代碼說話：

Java代碼

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));
}
}

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。