av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術(shù)文章
文章詳情頁

Java多線程之Disruptor入門

瀏覽:108日期:2022-08-13 13:31:52
一、Disruptor簡介

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

二、淺聊Disruptor的核心

Java多線程之Disruptor入門  

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

三、Disruptor使用3.1 pom.xml

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

Disruptor是基于事件的生產(chǎn)者消費(fèi)者模型。其RingBuffer中存放的其實(shí)是將消息封裝成的事件。這里定義了一個(gè)LongEvent,表示消息隊(duì)列中存放的是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

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

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

實(shí)現(xiàn)EventHandler接口,定義EventHandler(消費(fèi)者),處理容器中的元素。

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;/** * 定義一個(gè)生產(chǎn)者,往Disruptor中投遞消息 */public class LongEventProducer { private RingBuffer<LongEvent> ringBuffer; public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {this.ringBuffer = ringBuffer; } public void onData(ByteBuffer byteBuffer) {// 定位到下一個(gè)可存放的位置long sequence = ringBuffer.next();try { // 拿到該位置的event LongEvent event = ringBuffer.get(sequence); // 設(shè)置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;// 構(gòu)造一個(gè)DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, Executors.defaultThreadFactory());// 綁定handlerdisruptor.handleEventsWith(new LongEventHandler());// 啟動(dòng)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);} }}

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

標(biāo)簽: Java
相關(guān)文章:
主站蜘蛛池模板: 精品色 | 久久精品中文字幕 | 久久久久国产一区二区三区四区 | 蜜月aⅴ免费一区二区三区 99re在线视频 | 黄在线免费观看 | 可以看黄的视频 | 中文字幕精 | 欧美黄色片 | 国产日韩视频 | 天天干在线播放 | 国产九九精品 | 国产一区二区三区在线 | 成人免费视频网站在线观看 | 亚洲三级视频 | 精品美女久久久久久免费 | 久久极品 | 久久国产成人 | 在线欧美小视频 | 高清欧美性猛交 | 国产精品久久久久一区二区三区 | 天天爽一爽 | 欧美日韩国产一区二区三区 | 成人免费毛片片v | 国产精品久久久久久久久久久久冷 | 欧美久久免费观看 | 日本在线小视频 | 欧美日韩久久精品 | 青青草华人在线视频 | 黑人巨大精品欧美一区二区免费 | 亚洲在线 | 国产精品无码永久免费888 | 91精品免费视频 | 黄色免费在线观看网站 | 人干人人| 一区二区三区精品视频 | 欧美色性 | 日韩网| 精品免费国产一区二区三区 | 中文字幕在线网 | 欧美性受xxxx白人性爽 | 九色视频网站 |