作者：霽云HYY

來源：https://blog.csdn.net/LO_YUN/article/details/103949317

• 生產(chǎn)者產(chǎn)生消息發(fā)送給RocketMQ
• RocketMQ接收到了消息之后，必然需要存到磁盤中，否則斷電或宕機(jī)之后會造成數(shù)據(jù)的丟失
• 消費(fèi)者從RocketMQ中獲取消息消費(fèi)，消費(fèi)成功之后，整個流程結(jié)束

• RocketMQ為了減少磁盤的IO，會先將消息寫入到os cache中，而不是直接寫入到磁盤中，消費(fèi)者從os cache中獲取消息類似于直接從內(nèi)存中獲取消息，速度更快，過一段時間會由os線程異步的將消息刷入磁盤中，此時才算真正完成了消息的持久化。在這個過程中，如果消息還沒有完成異步刷盤，RocketMQ中的Broker宕機(jī)的話，就會導(dǎo)致消息丟失
• 如果消息已經(jīng)被刷入了磁盤中，但是數(shù)據(jù)沒有做任何備份，一旦磁盤損壞，那么消息也會丟失

• 首先生產(chǎn)者發(fā)送half消息到RocketMQ中，此時消費(fèi)者是無法消費(fèi)half消息的，若half消息就發(fā)送失敗了，則執(zhí)行相應(yīng)的回滾邏輯
• half消息發(fā)送成功之后，且RocketMQ返回成功響應(yīng)，則執(zhí)行生產(chǎn)者的核心鏈路
• 如果生產(chǎn)者自己的核心鏈路執(zhí)行失敗，則回滾，并通知RocketMQ刪除half消息
• 如果生產(chǎn)者的核心鏈路執(zhí)行成功，則通知RocketMQ commit half消息，讓消費(fèi)者可以消費(fèi)這條數(shù)據(jù)

//注冊消息監(jiān)聽器處理消息
consumer.registerMessageListener(new?MessageListenerConcurrently()?{
???@Override
????public?ConsumeConcurrentlyStatus?consumeMessage(List?msgs,?ConsumeConcurrentlyContext?context){??????????????????????????????????
????????//對消息進(jìn)行處理
????????return?ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
????}
});

//注冊消息監(jiān)聽器處理消息
consumer.registerMessageListener(new?MessageListenerConcurrently()?{
???@Override
????public?ConsumeConcurrentlyStatus?consumeMessage(List?msgs,?ConsumeConcurrentlyContext?context){?
?????//開啟子線程異步處理消息
?????new?Thread()?{
???public?void?run()?{
????//對消息進(jìn)行處理
???}
??}.start();?????????????????????????????????
????????return?ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
????}
});

• 使用事務(wù)機(jī)制傳輸消息，會比普通的消息傳輸多出很多步驟，耗費(fèi)性能
• 同步刷盤相比異步刷盤，一個是存儲在磁盤中，一個存儲在內(nèi)存中，速度完全不是一個數(shù)量級
• 主從機(jī)構(gòu)的話，需要Leader將數(shù)據(jù)同步給Follower
• 消費(fèi)時無法異步消費(fèi)，只能等待消費(fèi)完成再通知RocketMQ消費(fèi)完成

特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：

長按訂閱更多精彩▼
如有收獲，點(diǎn)個在看，誠摯感謝