Disruptor目前是世界上最快的單機消息隊列，由英國外匯交易公司LMAX開發(fā)，研發(fā)的初衷是解決內(nèi)存隊列的延遲問題（在性能測試中發(fā)現(xiàn)竟然與I/O操作處于同樣的數(shù)量級）。基于Disruptor開發(fā)的系統(tǒng)單線程能支撐每秒600萬訂單，2010年在QCon演講后，獲得了業(yè)界關注。2011年，企業(yè)應用軟件專家Martin Fowler專門撰寫長文介紹。同年它還獲得了Oracle官方的Duke大獎。目前，包括Apache Storm、Camel、Log4j 2在內(nèi)的很多知名項目都應用了Disruptor以獲取高性能。

Disruptor維護了一個環(huán)形隊列RingBuffer，這個隊列本質(zhì)上是一個首位相連的數(shù)組。相比于LinkedBlockdingQueue，RingBuffer的數(shù)組結(jié)構在查找方面效率更高。此外，LinkedBlockingQueue需要維護一個頭節(jié)點指針head和一個尾節(jié)點指針tail，而RingBuffer只需要維護一個sequence指向下一個可用的位置即可。所以從這兩點來說，RingBuffer比LinkedBlockingQueue要快。

<dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.4.3</version></dependency>3.2 事件Event

Disruptor是基于事件的生產(chǎn)者消費者模型。其RingBuffer中存放的其實是將消息封裝成的事件。這里定義了一個LongEvent，表示消息隊列中存放的是long類型的數(shù)據(jù)。

public class LongEvent {private long value;public void set(long value) {this.value = value;} @Override public String toString() {return 'LongEvent{' +'value=' + value +’}’; }}3.3 EventFactory

實現(xiàn)EventFactory接口，定義Event工廠，用于填充隊列。Event工廠其實是為了提高Disruptor的效率，初始化的時候，會調(diào)用Event工廠，對RingBuffer進行內(nèi)存的提前分配，GC的頻率會降低。

import com.lmax.disruptor.EventFactory;public class LongEventFactory implements EventFactory<LongEvent> {public LongEvent newInstance() {return new LongEvent();}}3.4 EventHandler

實現(xiàn)EventHandler接口，定義EventHandler(消費者)，處理容器中的元素。

import com.lmax.disruptor.EventHandler;public class LongEventHandler implements EventHandler<LongEvent> {public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {System.out.println('Event: ' + event + ', sequence: ' + sequence);}}3.5 使用Disruptor原始API發(fā)布消息

import cn.flying.space.disruptor.demo.LongEvent;import com.lmax.disruptor.RingBuffer;import java.nio.ByteBuffer;/** * 定義一個生產(chǎn)者，往Disruptor中投遞消息 */public class LongEventProducer { private RingBuffer<LongEvent> ringBuffer; public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {this.ringBuffer = ringBuffer; } public void onData(ByteBuffer byteBuffer) {// 定位到下一個可存放的位置long sequence = ringBuffer.next();try { // 拿到該位置的event LongEvent event = ringBuffer.get(sequence); // 設置event的值 event.set(byteBuffer.getLong(0));} finally { // 發(fā)布 ringBuffer.publish(sequence);} }}import cn.flying.space.disruptor.demo.LongEvent;import cn.flying.space.disruptor.demo.LongEventFactory;import cn.flying.space.disruptor.demo.LongEventHandler;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.dsl.Disruptor;import java.nio.ByteBuffer;import java.util.concurrent.Executors;public class TestMain { public static void main(String[] args) throws InterruptedException {// 定義event工廠LongEventFactory factory = new LongEventFactory();// ringBuffer長度int bufferSize = 1024;// 構造一個DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, Executors.defaultThreadFactory());// 綁定handlerdisruptor.handleEventsWith(new LongEventHandler());// 啟動Disruptordisruptor.start();RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();LongEventProducer producer = new LongEventProducer(ringBuffer);ByteBuffer byteBuffer = ByteBuffer.allocate(8);for (long i = 0; true; i++) { byteBuffer.clear(); byteBuffer.putLong(i); // 投遞消息 producer.onData(byteBuffer); Thread.sleep(1000);} }}3.6 使用Translators發(fā)布消息

import cn.flying.space.disruptor.demo.LongEvent;import com.lmax.disruptor.EventTranslatorOneArg;import com.lmax.disruptor.RingBuffer;import java.nio.ByteBuffer;public class LongEventProducerUsingTranslator { private RingBuffer<LongEvent> ringBuffer; public LongEventProducerUsingTranslator(RingBuffer<LongEvent> ringBuffer) {this.ringBuffer = ringBuffer; } private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR = new EventTranslatorOneArg<LongEvent, ByteBuffer>() {@Overridepublic void translateTo(LongEvent longEvent, long l, ByteBuffer byteBuffer) { longEvent.set(byteBuffer.getLong(0));} }; public void onData(ByteBuffer byteBuffer) {ringBuffer.publishEvent(TRANSLATOR, byteBuffer); }}import cn.flying.space.disruptor.demo.LongEvent;import cn.flying.space.disruptor.demo.LongEventFactory;import cn.flying.space.disruptor.demo.LongEventHandler;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;/** * @author ZhangSheng * @date 2021-4-26 14:23 */public class TestMain { public static void main(String[] args) throws InterruptedException {LongEventFactory factory = new LongEventFactory();int bufferSize = 1024;Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);disruptor.handleEventsWith(new LongEventHandler());disruptor.start();RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();LongEventProducerUsingTranslator producer = new LongEventProducerUsingTranslator(ringBuffer);ByteBuffer byteBuffer = ByteBuffer.allocate(8);for (long i = 0L; true; i++) { byteBuffer.putLong(0, i); // 發(fā)布 producer.onData(byteBuffer); Thread.sleep(1000);} }}

到此這篇關于Java多線程之Disruptor入門的文章就介紹到這了,更多相關Java Disruptor入門內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持好吧啦網(wǎng)！