順序訊息(FIFO訊息)是雲訊息佇列 RocketMQ 版提供的一種嚴格按照順序來發布和消費的訊息類型。本文提供使用TCP協議下的Java SDK收發順序訊息的範例程式碼。
前提條件
您已完成以下操作:
背景資訊
順序訊息分為兩類:
全域順序:對於指定的一個Topic,所有訊息按照嚴格的先入先出FIFO(First In First Out)的順序進行發布和消費。
分區順序:對於指定的一個Topic,所有訊息根據Sharding Key進行區塊分區。同一個分區內的訊息按照嚴格的FIFO順序進行發布和消費。Sharding Key是順序訊息中用來區分不同分區的關鍵字段,和普通訊息的Key是完全不同的概念。
更多資訊,請參見順序訊息。
說明
對於新手使用者,建議在正式收發訊息前,閱讀Demo工程來瞭解搭建雲訊息佇列 RocketMQ 版工程的具體步驟。
發送順序訊息
重要
雲訊息佇列 RocketMQ 版服務端判定訊息產生的順序性是參照單一生產者、單一線程並發下訊息發送的時序。如果發送方有多個生產者或者有多個線程並發發送訊息,則此時只能以到達雲訊息佇列 RocketMQ 版服務端的時序作為訊息順序的依據,和業務側的發送順序未必一致。
具體的範例程式碼,請以雲訊息佇列 RocketMQ 版程式碼程式庫為準。
全域順序訊息和分區順序訊息的發布方式基本一樣,範例程式碼如下。
package com.aliyun.openservices.ons.example.order;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import java.util.Properties;
public class ProducerClient {
public static void main(String[] args) {
Properties properties = new Properties();
// 您在訊息佇列RocketMQ版控制台建立的Group ID。
properties.put(PropertyKeyConst.GROUP_ID,"XXX");
// AccessKey ID,阿里雲身分識別驗證標識。
properties.put(PropertyKeyConst.AccessKey,"XXX");
// AccessKey Secret,阿里雲身分識別驗證密鑰。
properties.put(PropertyKeyConst.SecretKey,"XXX");
// 設定TCP接入網域名稱,進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXX");
OrderProducer producer = ONSFactory.createOrderProducer(properties);
// 在發送訊息前,必須調用start方法來啟動Producer,只需調用一次即可。
producer.start();
for (int i = 0; i < 1000; i++) {
String orderId = "biz_" + i % 10;
Message msg = new Message(
// Message所屬的Topic。
"Order_global_topic",
// Message Tag,可理解為Gmail中的標籤,對訊息進行再歸類,方便Consumer指定過濾條件在訊息佇列RocketMQ版的伺服器過濾。
"TagA",
// Message Body,可以是任何二進位形式的資料,訊息佇列RocketMQ版不做任何幹預,需要Producer與Consumer協商好一致的序列化和還原序列化方式。
"send order global msg".getBytes()
);
// 設定代表訊息的業務關鍵屬性,請儘可能全域唯一。
// 以方便您在無法正常收到訊息情況下,可通過訊息佇列RocketMQ版控制台查詢訊息並補發。
// 注意:不設定也不會影響訊息正常收發。
msg.setKey(orderId);
// 分區順序訊息中區分不同分區的關鍵字段,Sharding Key與普通訊息的key是完全不同的概念。
// 全域順序訊息,該欄位可以設定為任意非Null 字元串。
String shardingKey = String.valueOf(orderId);
try {
SendResult sendResult = producer.send(msg, shardingKey);
// 發送訊息,只要不拋異常就是成功。
if (sendResult != null) {
System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
}
}
catch (Exception e) {
// 訊息發送失敗,需要進行重試處理,可重新發送這條訊息或持久化這條資料進行補償處理。
System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
e.printStackTrace();
}
}
// 在應用退出前,銷毀Producer對象。
// 注意:銷毀Producer對象可以節約系統記憶體,若您需要頻繁發送訊息,則無需銷毀Producer對象。
producer.shutdown();
}
}
訂閱順序訊息
全域順序訊息和分區順序訊息的訂閱者式基本一樣,範例程式碼如下。
package com.aliyun.openservices.ons.example.order;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import java.util.Properties;
public class ConsumerClient {
public static void main(String[] args) {
Properties properties = new Properties();
// 您在訊息佇列RocketMQ版控制台建立的Group ID。
properties.put(PropertyKeyConst.GROUP_ID,"XXX");
// AccessKey ID,阿里雲身分識別驗證標識。
properties.put(PropertyKeyConst.AccessKey,"XXX");
// AccessKey Secret,阿里雲身分識別驗證密鑰。
properties.put(PropertyKeyConst.SecretKey,"XXX");
// 設定TCP接入網域名稱,進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXX");
// 順序訊息消費失敗進行重試前的等待時間,單位(毫秒),取值範圍: 10毫秒~30,000毫秒。
properties.put(PropertyKeyConst.SuspendTimeMillis,"100");
// 訊息消費失敗時的最大重試次數。
properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");
// 在訂閱訊息前,必須調用start方法來啟動Consumer,只需調用一次即可。
OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);
consumer.subscribe(
// Message所屬的Topic。
"Order_global_topic",
// 訂閱指定Topic下的Tags:
// 1. * 表示訂閱所有訊息。
// 2. TagA || TagB || TagC表示訂閱TagA或TagB或TagC的訊息。
"*",
new MessageOrderListener() {
/**
* 1. 訊息消費處理失敗或者處理出現異常,返回OrderAction.Suspend。
* 2. 訊息處理成功,返回OrderAction.Success。
*/
@Override
public OrderAction consume(Message message, ConsumeOrderContext context) {
System.out.println(message);
return OrderAction.Success;
}
});
consumer.start();
}
}