自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

三分鐘白話RocketMQ系列—— 如何發(fā)送消息

開(kāi)發(fā) 前端
當(dāng)消息發(fā)送者向某個(gè) Broker 發(fā)送消息后,如果返回異常,生產(chǎn)者會(huì)在接下來(lái)的一段時(shí)間內(nèi)(例如5分鐘)避免再次選擇該 Broker 上的隊(duì)列來(lái)發(fā)送消息。這樣做的目的是規(guī)避可能發(fā)生故障的 Broker。

我們知道RocketMQ主要分為消息 生產(chǎn)、存儲(chǔ)(消息堆積)、消費(fèi) 三大塊領(lǐng)域。

那接下來(lái),我們白話一下,RocketMQ是如何發(fā)送消息的,揭秘消息生產(chǎn)全過(guò)程。

注意,如果白話中不小心提到相關(guān)代碼配置與類名,請(qǐng)參考RocketMQ 4.9.4版本

關(guān)鍵字摘要

  • 哪些消息類型?
  • 發(fā)給誰(shuí)?
  • 怎么發(fā)?
  • 怎么知道發(fā)成功了還是失敗了?
  • 發(fā)失敗了怎么辦?

Q1: RocketMQ有哪些消息類型?

RocketMQ生產(chǎn)消息時(shí),支持多種「消息類型」:

  • 普通消息:發(fā)送普通消息。
SendResult send(final Message msg);
  • 普通有序消息:發(fā)送普通有序消息,通過(guò)指定「消息篩選器selector」,動(dòng)態(tài)決定發(fā)送哪個(gè)隊(duì)列。
SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg);
  • 嚴(yán)格有序消息:發(fā)送嚴(yán)格有序消息,通過(guò)指定隊(duì)列,保證嚴(yán)格有序。
SendResult send(final Message msg, final MessageQueue mq);
  • 事務(wù)消息:實(shí)現(xiàn)分布式事務(wù)。(屬于分布式事務(wù)范疇,區(qū)別較大,本文不再展開(kāi)討論,后面單獨(dú)寫(xiě)一篇針對(duì)「事務(wù)消息」的分析)

上面列舉的三種send方法,都是以同步發(fā)送模式為例。定時(shí)/延遲消息從發(fā)送方式角度來(lái)說(shuō),不算一種獨(dú)立的消息類型。

Q2:RocketMQ怎么知道一條消息要發(fā)送給誰(shuí)?

一般我們要發(fā)送一條消息給RocketMQ,需要?jiǎng)?chuàng)建這樣一個(gè)消息體。

Message msg = new Message( "TestTopic", "Hello World".getBytes() );

在這個(gè)消息體里面,我們只單純指定了要發(fā)送的Topic名字,以及要發(fā)送的消息內(nèi)容。

那么,RocketMQ-client怎么知道這條消息要發(fā)送到RocketMQ集群中的哪一個(gè)broker上呢?

這里需要了解下RocketMQ中Topic的「路由注冊(cè)與發(fā)現(xiàn)機(jī)制」。

RocketMQ基本架構(gòu)RocketMQ基本架構(gòu)

Topic 路由注冊(cè)與發(fā)現(xiàn):

  • Broker 每30秒向 NameServer 發(fā)送心跳包,里面包含Topic的路由信息,包括主題的讀寫(xiě)隊(duì)列數(shù)和操作權(quán)限等。NameServer會(huì)保存這些路由信息,并記錄最后一次收到 Broker 心跳包的時(shí)間(NameServer每10秒根據(jù)記錄的時(shí)間戳清理已經(jīng)失聯(lián)120秒以上的 Broker)。
  • 生產(chǎn)者每30秒獲取一次主題的路由信息。這意味著消息生產(chǎn)者不會(huì)立即知道有新的 Broker 加入或者被移除。

Topic路由信息Topic路由信息

Topic的路由信息,包括了Topic的 隊(duì)列queue和broker的映射關(guān)系 ,那么如何利用這個(gè)Topic的路由信息呢?

我們需要根據(jù)前面的不同「消息類型」進(jìn)行分別討論:

  • 普通消息:默認(rèn)采用輪詢機(jī)制,消息會(huì)依次發(fā)送到Topic的每個(gè)可用的 Broker 的某個(gè)隊(duì)列queue上,以實(shí)現(xiàn)負(fù)載均衡。
  • 普通有序消息:根據(jù)傳遞的 MessageQueueSelector 和消息體 msg 內(nèi)容,計(jì)算可以投遞的隊(duì)列queue,然后發(fā)送消息。(可以類比分庫(kù)分表中的分表計(jì)算寫(xiě)入的方式)
  • 嚴(yán)格有序消息:根據(jù)傳遞的 MessageQueue 信息,強(qiáng)制消息發(fā)送到對(duì)應(yīng)隊(duì)列queue上。(可以類比分庫(kù)分表中,強(qiáng)制指定物理表寫(xiě)入的方式)

根據(jù)消息類型獲取到目標(biāo)隊(duì)列queue后,就可以根據(jù)Topic路由信息發(fā)送消息到指定broker上了。

Q3:怎么發(fā)送一條消息?

從發(fā)送模式角度來(lái)說(shuō),RocketMQ有三種「消息發(fā)送模式」:

  • 同步發(fā)送:調(diào)用發(fā)送消息方法后,同步阻塞,直到返回SendResult。
SendResult send(final Message msg);
  • 異步發(fā)送:調(diào)用發(fā)送消息方法后,立即返回,發(fā)送結(jié)果會(huì)通過(guò)開(kāi)發(fā)者自己注冊(cè)的回調(diào)函數(shù)SendCallback進(jìn)行處理。
void send(final Message msg, final SendCallback sendCallback);
  • 單向發(fā)送:這種方法完全不關(guān)心發(fā)送后的返回結(jié)果。顯然,它具有最大吞吐量,但也存在消息丟失的潛在風(fēng)險(xiǎn)。
void sendOneway(final Message msg);

上面列舉的三種send方法,都是以「普通消息」為例。

「消息類型」 和 「消息發(fā)送模式」 是 N*M 的關(guān)系,所以聰明的你一定已經(jīng)想到了,存在9種不同組合(不包括事物消息),RocketMQ也是在接口中定義了9種不同方法。

Q4: 發(fā)送后,怎么知道消息發(fā)成功了還是失敗了?

前面介紹了三種「消息發(fā)送模式」,其中「單向發(fā)送」屬于不可靠發(fā)送,我們無(wú)法知道是否發(fā)送成功。

而「同步發(fā)送」和「異步發(fā)送」都是可靠發(fā)送,我們能夠獲取發(fā)送狀態(tài),知道是否成功。

在「同步發(fā)送」中,我們可以根據(jù)SendResult中的sendStatus屬性判斷是否發(fā)送成功。

SendResult類屬性SendResult類屬性

在「異步發(fā)送」中,我們可以自定義實(shí)現(xiàn)SendCallback的onSuccess()方法和onException()方法,來(lái)判斷消息是否發(fā)送成功。

SendCallback接口定義SendCallback接口定義

Q5: 消息發(fā)送失敗了怎么辦?

如果消息發(fā)送失敗了,RocketMQ-client默認(rèn)有重試機(jī)制,以確保消息的高可用性。

前面提到,生產(chǎn)者每30秒獲取一次主題的路由信息,所以即使某個(gè) Broker 宕機(jī),消息發(fā)送者可能無(wú)法立即察覺(jué)到它的宕機(jī)狀態(tài)。

但是,當(dāng)消息發(fā)送者向某個(gè) Broker 發(fā)送消息后,如果返回異常,生產(chǎn)者會(huì)在接下來(lái)的一段時(shí)間內(nèi)(例如5分鐘)避免再次選擇該 Broker 上的隊(duì)列來(lái)發(fā)送消息。這樣做的目的是規(guī)避可能發(fā)生故障的 Broker。

當(dāng)然了,用戶也能根據(jù)返回的異常,自己定義業(yè)務(wù)重試、補(bǔ)償機(jī)制。

需要注意的是,不同「消息類型」和「消息發(fā)送模式」的RocketMQ-client默認(rèn)重試機(jī)制不同。

消息類型:

  • 普通消息:無(wú)順序性要求,異常時(shí)RocketMQ-client默認(rèn)重試。
  • 普通有序消息:異常時(shí)RocketMQ-client默認(rèn)不重試,可以用戶自己捕獲異常重試,并發(fā)送到其他隊(duì)列。
  • 嚴(yán)格有序消息:保證嚴(yán)格有序,異常時(shí)RocketMQ-client默認(rèn)不重試,可以用戶自己捕獲異常重試。

注意:有序消息異常時(shí)RocketMQ-client都是默認(rèn)不重試

消息發(fā)送模式:

  • 同步發(fā)送:配置retryTimesWhenSendFailed默認(rèn)重試次數(shù)。
  • 異步發(fā)送:配置retryTimesWhenSendAsyncFailed默認(rèn)重試次數(shù)。
  • 單向發(fā)送:無(wú)重試機(jī)制,存在丟失消息的風(fēng)險(xiǎn)。

注意:?jiǎn)蜗虬l(fā)送模式異常時(shí)RocketMQ-client默認(rèn)不重試

總結(jié)

  • 有哪些消息類型:普通消息、有序消息、事務(wù)消息
  • 發(fā)給誰(shuí)?:Topic路由信息注冊(cè)與發(fā)現(xiàn)機(jī)制、普通消息輪詢發(fā)送、有序消息指定selector或者queue發(fā)送
  • 怎么發(fā)?:同步發(fā)送、異步發(fā)送、單向發(fā)送
  • 怎么知道發(fā)成功了還是失敗了?:同步&異步都能夠獲取發(fā)送狀態(tài)(可靠發(fā)送)、單向發(fā)送不可靠
  • 發(fā)失敗了怎么辦?: 失敗重試機(jī)制
責(zé)任編輯:武曉燕 來(lái)源: 阿丸筆記
相關(guān)推薦

2023-08-24 09:01:25

消息拉取RocketMQ

2023-08-14 09:16:26

消息存儲(chǔ)磁盤(pán)

2023-09-13 08:14:57

RocketMQ次數(shù)機(jī)制

2023-09-21 09:02:03

RocketMQ全局有序局部有序

2023-07-25 09:00:27

RocketMQ開(kāi)源

2024-04-01 09:59:08

消息隊(duì)列通信微服務(wù)

2024-09-13 08:49:45

2024-05-16 11:13:16

Helm工具release

2009-11-09 12:55:43

WCF事務(wù)

2024-12-18 10:24:59

代理技術(shù)JDK動(dòng)態(tài)代理

2022-02-17 09:24:11

TypeScript編程語(yǔ)言javaScrip

2021-04-20 13:59:37

云計(jì)算

2023-12-27 08:15:47

Java虛擬線程

2024-01-16 07:46:14

FutureTask接口用法

2024-08-30 08:50:00

2013-06-28 14:30:26

棱鏡計(jì)劃棱鏡棱鏡監(jiān)控項(xiàng)目

2020-06-30 10:45:28

Web開(kāi)發(fā)工具

2021-12-17 07:47:37

IT風(fēng)險(xiǎn)框架

2024-10-15 09:18:30

2024-01-12 07:38:38

AQS原理JUC
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)