Java多線程之Disruptor入門

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

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

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

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

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

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

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

實現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發布消息

import cn.flying.space.disruptor.demo.LongEvent;import com.lmax.disruptor.RingBuffer;import java.nio.ByteBuffer;/** * 定義一個生產者，往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 { // 發布 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發布消息

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); // 發布 producer.onData(byteBuffer); Thread.sleep(1000);} }}

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