• 普通消息

• 定時和延時消息

• 順序消息

• 事務(wù)消息

普通消息

普通消息是指消息隊列中無特性的消息，區(qū)別于有特性的定時/延時消息、順序消息和事務(wù)消息。

TCP Java SDK 收發(fā)普通消息的示例代碼

• 發(fā)送普通消息（三種方式）

• 發(fā)送消息（多線程）

• 訂閱消息

定時和延時消息

• 定時消息：Producer 將消息發(fā)送到消息隊列服務(wù)端，但并不期望這條消息立馬投遞，而是推遲到在當(dāng)前時間點之后的某一個時間投遞到 Consumer 進行消費，該消息即定時消息。

• 延時消息：Producer 將消息發(fā)送到消息隊列服務(wù)端，但并不期望這條消息立馬投遞，而是延遲一定時間后才投遞到 Consumer 進行消費，該消息即延時消息。

適用場景

定時消息和延時消息適用于以下場景：

消息生產(chǎn)和消費有時間窗口要求：比如在螞蟻森林場景中，相關(guān)低碳行為觸發(fā)時，會發(fā)送一條定時消息，在第二天 7 點定時投遞給消費者，產(chǎn)生綠色能量；或者發(fā)送一條延時消息，24 小時后產(chǎn)生綠色能量。

使用方式

定時消息和延時消息的使用在代碼編寫上存在略微的區(qū)別：

• 發(fā)送定時消息需要明確指定消息發(fā)送時間點之后的某一時間點作為消息投遞的時間點。

• 發(fā)送延時消息時需要設(shè)定一個延時時間長度，消息將從當(dāng)前發(fā)送時間點開始延遲固定時間之后才開始投遞。

注意事項

• 定時和延時消息的 msg.setStartDeliverTime 參數(shù)需要設(shè)置成當(dāng)前時間戳之后的某個時刻（單位毫秒）。如果被設(shè)置成當(dāng)前時間戳之前的某個時刻，消息將立刻投遞給消費者。

• 定時和延時消息的 msg.setStartDeliverTime 參數(shù)可設(shè)置 3 天內(nèi)的任何時刻（單位毫秒），超過 3 天消息發(fā)送將失敗。

• StartDeliverTime 是服務(wù)端開始向消費端投遞的時間。 如果消費者當(dāng)前有消息堆積，那么定時和延時消息會排在堆積消息后面，將不能嚴(yán)格按照配置的時間進行投遞。

• 由于客戶端和服務(wù)端可能存在時間差，消息的實際投遞時間與客戶端設(shè)置的投遞時間之間可能存在偏差。

• 設(shè)置定時和延時消息的投遞時間后，依然受 3 天的消息保存時長限制。

TCP 協(xié)議示例代碼

收發(fā)定時消息和延時消息的示例代碼（Java），請參見 收發(fā)延時消息。

順序消息

順序消息（FIFO 消息）是消息隊列提供的一種嚴(yán)格按照順序來發(fā)布和消費的消息。順序發(fā)布和順序消費是指對于指定的一個 Topic，生產(chǎn)者按照一定的先后順序發(fā)布消息；消費者按照既定的先后順序訂閱消息，即先發(fā)布的消息一定會先被客戶端接收到。

順序消息目前支持分區(qū)順序消息。

分區(qū)順序消息

對于指定的一個 Topic，所有消息根據(jù) Sharding Key 進行區(qū)塊分區(qū)。同一個分區(qū)內(nèi)的消息按照嚴(yán)格的 FIFO 順序進行發(fā)布和消費。Sharding Key 是順序消息中用來區(qū)分不同分區(qū)的關(guān)鍵字段，和普通消息的 Key 是完全不同的概念。

• 適用于性能要求高，以 Sharding Key 作為分區(qū)字段，在同一個區(qū)塊中嚴(yán)格地按照 FIFO 原則進行消息發(fā)布和消費的場景。

• 示例說明如下：

  用戶注冊需要發(fā)送發(fā)驗證碼，以用戶 ID 作為 Sharding Key，那么同一個用戶發(fā)送的消息都會按照發(fā)布的先后順序來消費。

注意事項

使用順序消息時，請注意以下幾點：

• 順序消息暫不支持廣播模式。

• 建議同一個 Group ID 只對應(yīng)一種類型的 Topic，即不同時用于順序消息和無序消息的收發(fā)。

• 順序消息不支持異步發(fā)送方式，否則將無法嚴(yán)格保證順序。

TCP SDK 示例代碼

TCP 協(xié)議下的示例代碼請參見 收發(fā)順序消息。

事務(wù)消息

• 事務(wù)消息：消息隊列提供類似 X/Open XA 的分布式事務(wù)功能，通過消息隊列事務(wù)消息能達到分布式事務(wù)的最終一致。

• 半事務(wù)消息：暫不能投遞的消息，發(fā)送方已經(jīng)成功地將消息發(fā)送到了消息隊列服務(wù)端，但是服務(wù)端未收到生產(chǎn)者對該消息的二次確認(rèn)，此時該消息被標(biāo)記成“暫不能投遞”狀態(tài)，處于該種狀態(tài)下的消息即半事務(wù)消息。

• 消息回查：由于網(wǎng)絡(luò)閃斷、生產(chǎn)者應(yīng)用重啟等原因，導(dǎo)致某條事務(wù)消息的二次確認(rèn)丟失，消息隊列服務(wù)端通過掃描發(fā)現(xiàn)某條消息長期處于“半事務(wù)消息”時，需要主動向消息生產(chǎn)者詢問該消息的最終狀態(tài)（Commit 或 Rollback），該詢問過程即消息回查。

適用場景

在轉(zhuǎn)賬過程中，比如從支付寶轉(zhuǎn)賬到余額寶，兩個系統(tǒng)之間的數(shù)據(jù)需要保持最終一致性，這時可以通過事務(wù)消息進行處理。支付寶扣款執(zhí)行前，發(fā)送一條半事務(wù)消息，扣款事務(wù)執(zhí)行成功后，將消息狀態(tài)更新為 Commit，余額寶系統(tǒng)訂閱消息隊列的扣款消息，做相應(yīng)的存款業(yè)務(wù)處理。

交互流程

事務(wù)消息交互流程如下圖。

事務(wù)消息發(fā)送步驟如下：

1. 發(fā)送方將半事務(wù)消息發(fā)送至消息隊列服務(wù)端。

2. 消息隊列服務(wù)端將消息持久化成功之后，向發(fā)送方返回 Ack 確認(rèn)消息已經(jīng)發(fā)送成功，此時消息為半事務(wù)消息。

3. 發(fā)送方開始執(zhí)行本地事務(wù)邏輯。

4. 發(fā)送方根據(jù)本地事務(wù)執(zhí)行結(jié)果向服務(wù)端提交二次確認(rèn)（Commit 或 Rollback），服務(wù)端收到 Commit 狀態(tài)則將半事務(wù)消息標(biāo)記為可投遞，訂閱方最終將收到該消息；服務(wù)端收到 Rollback 狀態(tài)則刪除半事務(wù)消息，訂閱方將不會接受該消息。

事務(wù)消息回查步驟如下：

1. 在斷網(wǎng)或者是應(yīng)用重啟的特殊情況下，上述步驟 4 提交的二次確認(rèn)最終未到達服務(wù)端，經(jīng)過固定時間后服務(wù)端將對該消息發(fā)起消息回查。

2. 發(fā)送方收到消息回查后，需要檢查對應(yīng)消息的本地事務(wù)執(zhí)行的最終結(jié)果。

3. 發(fā)送方根據(jù)檢查得到的本地事務(wù)的最終狀態(tài)再次提交二次確認(rèn)，服務(wù)端仍按照步驟 4 對半事務(wù)消息進行操作。

注意事項

1. 事務(wù)消息的 Group ID 不能與其他類型消息的 Group ID 共用。與其他類型的消息不同，事務(wù)消息有回查機制，回查時消息隊列服務(wù)端會根據(jù) Group ID 去查詢客戶端。

2. 通過 AccessPoint.getAccessPoint().createTransactionProducer 創(chuàng)建事務(wù)消息的 Producer 時必須指定 LocalTransactionChecker 的實現(xiàn)類，處理異常情況下事務(wù)消息的回查。

3. 事務(wù)消息發(fā)送完成本地事務(wù)后，可在 execute 方法中返回以下三種狀態(tài)：

   • TransactionStatus.CommitTransaction：提交事務(wù)，允許訂閱方消費該消息。

   • TransactionStatus.RollbackTransaction：回滾事務(wù)，消息將被丟棄不允許消費。

   • TransactionStatus.Unknow：暫時無法判斷狀態(tài)，等待固定時間以后消息隊列服務(wù)端向發(fā)送方進行消息回查。

4. 可通過以下方式給每條消息設(shè)定第一次消息回查的最快時間：

   Message message =new Message();
   // 在消息屬性中添加第一次消息回查的最快時間，單位秒。例如，以下設(shè)置實際第一次回查時間為 120 秒 ~ 125 秒之間
   message.putUserProperties(PropertyKeyConst.CheckImmunityTimeInSeconds,"120");
   // 以上方式只確定事務(wù)消息的第一次回查的最快時間，實際回查時間向后浮動 0 秒 ~ 60 秒；如第一次回查后事務(wù)仍未提交，后續(xù)每隔 60 秒回查一次

TCP SDK 示例代碼

收發(fā)事務(wù)消息的示例代碼如下：

TCP Java SDK 收發(fā)事務(wù)消息