Spring Boot + Disruptor 實作內部訊息佇列

一、背景

工作中遇到計畫使用Disruptor做訊息佇列,對你沒看錯,不是Kafka,也不是rabbitmq;Disruptor有個最大的優點就是快,還有一點它是開源的哦,下面做個簡單的記錄.

二、Disruptor介紹

1、 Disruptor是英國外匯交易公司LMAX開發的一個高效能佇列，研發的初衷是解決記憶體佇列的延遲問題（在效能測試中發現竟然與I/O操作處於同樣的數量級）基於Disruptor開發的系統單執行緒能支撐每秒600萬訂單，2010年在QCon演講後，獲得了業界關註；
2、 Disruptor是一個開源的Java框架，它被設計用於在生產者—消費者（producer-consumerproblem，簡稱PCP）問題上獲得盡量高的吞吐量（TPS）和盡量低的延遲；
3、 從功能上來看，Disruptor是實作了「佇列」的功能，而且是一個有界佇列那麽它的套用場景自然就是「生產者-消費者」模型的套用場合了；
4、 Disruptor是LMAX線上交易平台的關鍵組成部份，LMAX平台使用該框架對訂單處理速度能達到600萬TPS，除金融領域之外，其他一般的套用中都可以用到Disruptor，它可以帶來顯著的效能提升；
5、 其實Disruptor與其說是一個框架，不如說是一種設計思路，這個設計思路對於存在「並行、緩沖區、生產者—消費者模型、事務處理」這些元素的程式來說，Disruptor提出了一種大幅提升效能（TPS）的方案；
6、 Disruptor的github主頁：https://github.com/LMAX-Exchange/disruptor；

三、Disruptor 的核心概念

先從了解 Disruptor 的核心概念開始，來了解它是如何運作的。下面介紹的概念模型，既是領域物件，也是對映到程式碼實作上的核心物件。

1. Ring Buffer

如其名，環形的緩沖區。曾經 RingBuffer 是 Disruptor 中的最主要的物件，但從3.0版本開始，其職責被簡化為僅僅負責對透過 Disruptor 進行交換的數據（事件）進行儲存和更新。在一些更高級的套用場景中，Ring Buffer 可以由使用者的自訂實作來完全替代。

2. Sequence Disruptor

透過順序遞增的序號來編號管理透過其進行交換的數據（事件），對數據(事件)的處理過程總是沿著序號逐個遞增處理。一個 Sequence 用於跟蹤標識某個特定的事件處理者( RingBuffer/Consumer )的處理進度。雖然一個 AtomicLong 也可以用於標識進度，但定義 Sequence 來負責該問題還有另一個目的，那就是防止不同的 Sequence 之間的CPU緩存偽共享(Flase Sharing)問題。

註：這是 Disruptor 實作高效能的關鍵點之一，網上關於偽共享問題的介紹已經汗牛充棟，在此不再贅述

3. Sequencer

Sequencer 是 Disruptor 的真正核心。此介面有兩個實作類 SingleProducerSequencer、MultiProducerSequencer ，它們定義在生產者和消費者之間快速、正確地傳遞數據的並行演算法。

4. Sequence Barrier

用於保持對RingBuffer的 main published Sequence 和Consumer依賴的其它Consumer的 Sequence 的參照。Sequence Barrier 還定義了決定 Consumer 是否還有可處理的事件的邏輯。

5. Wait Strategy

定義Consumer 如何進行等待下一個事件的策略。（註：Disruptor 定義了多種不同的策略，針對不同的場景，提供了不一樣的效能表現）

6. Event

在Disruptor 的語意中，生產者和消費者之間進行交換的數據被稱為事件(Event)。它不是一個被 Disruptor 定義的特定型別，而是由 Disruptor 的使用者定義並指定。

7. EventProcessor

EventProcessor 持有特定消費者(Consumer)的 Sequence，並提供用於呼叫事件處理實作的事件迴圈(Event Loop)。

8. EventHandler

Disruptor 定義的事件處理介面，由使用者實作，用於處理事件，是 Consumer 的真正實作。

9. Producer

即生產者，只是泛指呼叫 Disruptor 釋出事件的使用者程式碼，Disruptor 沒有定義特定介面或型別。

四、案例-demo

透過下面8個步驟,你就能將Disruptor Get回家啦：

1、添加pom.xml依賴

<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.4</version>
</dependency>


2、訊息體Model

/**
 * 訊息體
 */
@Data
public classMessageModel{

private String message;
}

3、構造EventFactory

public classHelloEventFactoryimplementsEventFactory<MessageModel> {

@Override
public MessageModel newInstance(){

returnnew MessageModel();
}
}


4、構造EventHandler-消費者

@Slf4j
public classHelloEventHandlerimplementsEventHandler<MessageModel> {

@Override
publicvoidonEvent(MessageModel event, long sequence, boolean endOfBatch){

try {

//這裏停止1000ms是為了確定消費訊息是異步的
Thread.sleep(1000);
log.info("消費者處理訊息開始");
if (event != null) {

log.info("消費者消費的資訊是：{}",event);
}
} catch (Exception e) {

log.info("消費者處理訊息失敗");
}
log.info("消費者處理訊息結束");
}
}





5、構造BeanManager

/**
 * 獲取例項化物件
 */
@Component
public classBeanManagerimplementsApplicationContextAware{
privatestatic ApplicationContext applicationContext = null;
@Override
publicvoidsetApplicationContext(ApplicationContext applicationContext)throws BeansException {

this.applicationContext = applicationContext;
}
publicstatic ApplicationContext getApplicationContext(){
return applicationContext; }
publicstatic Object getBean(String name){

return applicationContext.getBean(name);
}
publicstatic <T> T getBean( class<T> clazz){

return applicationContext.getBean(clazz);
}
}









6、構造MQManager

@Configuration
public classMQManager{
@Bean("messageModel")
public RingBuffer<MessageModel> messageModelRingBuffer(){

//定義用於事件處理的執行緒池， Disruptor透過java.util.concurrent.ExecutorSerivce提供的執行緒來觸發consumer的事件處理
ExecutorService executor = Executors.newFixedThreadPool(2);
//指定事件工廠
HelloEventFactory factory = new HelloEventFactory();
//指定ringbuffer字節大小，必須為2的N次方（能將求模運算轉為位運算提高效率），否則將影響效率
int bufferSize = 1024 * 256;
//單執行緒模式，獲取額外的效能
Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor,
ProducerType.SINGLE, new BlockingWaitStrategy());
//設定事件業務處理器---消費者
disruptor.handleEventsWith(new HelloEventHandler());
// 啟動disruptor執行緒
disruptor.start();
//獲取ringbuffer環，用於接取生產者生產的事件
RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();
return ringBuffer;
}









7、構造Mqservice和實作類-生產者

publicinterfaceDisruptorMqService{
/**
* 訊息
* @param message
*/
voidsayHelloMq(String message);
}
@Slf4j
@Component
@Service
public classDisruptorMqServiceImplimplementsDisruptorMqService{
@Autowired
private RingBuffer<MessageModel> messageModelRingBuffer;
@Override
publicvoidsayHelloMq(String message){

log.info("record the message: {}",message);
//獲取下一個Event槽的下標
long sequence = messageModelRingBuffer.next();
try {

//給Event填充數據
MessageModel event = messageModelRingBuffer.get(sequence);
event.setMessage(message);
log.info("往訊息佇列中添加訊息：{}", event);
} catch (Exception e) {

log.error("failed to add event to messageModelRingBuffer for : e = {},{}",e,e.getMessage());
} finally {

//釋出Event，啟用觀察者去消費，將sequence傳遞給改消費者
//註意最後的publish方法必須放在finally中以確保必須得到呼叫；如果某個請求的sequence未被送出將會堵塞後續的釋出操作或者其他的producer
messageModelRingBuffer.publish(sequence);
}
}
}







8、構造測試類及方法

@Slf4j
@RunWith(SpringRunner. class)
@SpringBootTest( classes= DemoApplication. class)
public classDemoApplicationTests{
@Autowired
private DisruptorMqService disruptorMqService;
/**
* 計畫內部使用Disruptor做訊息佇列
* @throws Exception
*/
@Test
publicvoidsayHelloMqTest()throws Exception{

disruptorMqService.sayHelloMq("訊息到了，Hello world!");
log.info("訊息佇列已發送完畢");
//這裏停止2000ms是為了確定是處理訊息是異步的
Thread.sleep(2000);
}
}


測試執行結果

2020-04-0514:31:18.543 INFO 7274 --- [ main] c.e.u.d.d.s.Impl.DisruptorMqServiceImpl : record the message: 訊息到了，Hello world!
2020-04-0514:31:18.545 INFO 7274 --- [ main] c.e.u.d.d.s.Impl.DisruptorMqServiceImpl : 往訊息佇列中添加訊息：MessageModel(message=訊息到了，Hello world!)
2020-04-0514:31:18.545 INFO 7274 --- [ main] c.e.utils.demo.DemoApplicationTests : 訊息佇列已發送完畢
2020-04-0514:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消費者處理訊息開始
2020-04-0514:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消費者消費的資訊是：MessageModel(message=訊息到了，Hello world!)
2020-04-0514:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消費者處理訊息結束

五、總結

其實生成者 -> 消費者 模式是很常見的，透過一些訊息佇列也可以輕松做到上述的效果。不同的地方在於，Disruptor 是在記憶體中以佇列的方式去實作的，而且是無鎖的。這也是 Disruptor 為什麽高效的原因。

如喜歡本文，請點選右上角，把文章分享到朋友圈
如有想了解學習的技術點，請留言給若飛安排分享

因公眾號更改推播規則，請點「在看」並加「星標」 第一時間獲取精彩技術分享

·END·

相關閱讀：

作者：不二天次

來源：blog.csdn.net/buertianci/article/details/105327031

版權申明：內容來源網路，僅供學習研究，版權歸原創者所有。如有侵權煩請告知，我們會立即刪除並表示歉意。謝謝!