在控制臺創建完所有資源之后,您可以調用云消息隊列 RocketMQ 版的TCP協議的SDK收發普通消息。
前提條件
- 說明
本文以普通消息為例進行說明,因此您創建的普通消息的Topic不能用來收發其他類型的消息,包括定時、延時、順序和事務消息。換言之,切勿混用不同消息類型的Topic。
下載并安裝TCP協議的SDK
說明
商業版SDK和社區版SDK相比,提供了更加豐富的功能特性并具有更高的穩定性保障,推薦您使用阿里云云消息隊列 RocketMQ 版商業版SDK進行接入。社區版SDK僅在開源RocketMQ遷移上云且不希望修改代碼時使用。
云消息隊列 RocketMQ 版提供了以下商業版TCP協議的SDK,請按需獲取相應語言的客戶端SDK。
調用TCP協議的SDK發送普通消息
獲取相應語言的客戶端SDK后,您即可運行以下示例代碼發送普通消息。
Java
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;
public class ProducerTest {
public static void main(String[] args) {
Properties properties = new Properties();
// AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
properties.put(PropertyKeyConst.AccessKey, "XXX");
// AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
properties.put(PropertyKeyConst.SecretKey, "XXX");
//設置發送超時時間,單位毫秒。
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
// 設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
properties.put(PropertyKeyConst.NAMESRV_ADDR,
"XXX");
Producer producer = ONSFactory.createProducer(properties);
// 在發送消息前,必須調用start方法來啟動Producer,只需調用一次即可。
producer.start();
Message msg = new Message(
// Message所屬的Topic。
"TopicTestMQ",
// Message Tag,可理解為Gmail中的標簽,對消息進行再歸類,方便Consumer指定過濾條件在消息隊列RocketMQ版的服務器過濾。
"TagA",
// Message Body,任何二進制形式的數據,消息隊列RocketMQ版不做任何干預,需要Producer與Consumer協商好一致的序列化和反序列化方式。
"Hello MQ".getBytes());
// 設置代表消息的業務關鍵屬性,請盡可能全局唯一。以方便您在無法正常收到消息情況下,可通過控制臺查詢消息并補發。
// 注意:不設置也不會影響消息正常收發。
msg.setKey("ORDERID_100");
// 異步發送消息,發送結果通過callback返回給客戶端。
producer.sendAsync(msg, new SendCallback() {
@Override
public void onSuccess(final SendResult sendResult) {
// 消息發送成功。
System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
}
@Override
public void onException(OnExceptionContext context) {
// 消息發送失敗,需要進行重試處理,可重新發送這條消息或持久化這條數據進行補償處理。
System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());
}
});
// 在callback返回之前即可取得msgId。
System.out.println("send message async. topic=" + msg.getTopic() + ", msgId=" + msg.getMsgID());
// 在應用退出前,銷毀Producer對象。 注意:如果不銷毀也沒有問題。
producer.shutdown();
}
}
.NET
using System;
using ons;
public class ProducerExampleForEx
{
public ProducerExampleForEx()
{
}
static void Main(string[] args) {
// 配置賬號,從控制臺獲取設置。
ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
// AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
// AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
// 您在控制臺創建的Group ID。
factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "GID_example");
// 您在控制臺創建的Topic。
factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
// 設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
// 設置日志路徑。
factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");
// 創建生產者實例。
// 說明:生產者實例是線程安全的,可用于發送不同Topic的消息。基本上,您每一個線程只需要一個生產者實例。
Producer producer = ONSFactory.getInstance().createProducer(factoryInfo);
// 啟動客戶端實例。
producer.start();
// 創建消息對象。
Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
msg.setKey(Guid.NewGuid().ToString());
for (int i = 0; i < 32; i++) {
try
{
SendResultONS sendResult = producer.send(msg);
Console.WriteLine("send success {0}", sendResult.getMessageId());
}
catch (Exception ex)
{
Console.WriteLine("send failure{0}", ex.ToString());
}
}
// 在您的線程即將退出時,關閉生產者實例。
producer.shutdown();
}
}
C/C++
#include "ONSFactory.h"
#include "ONSClientException.h"
using namespace ons;
int main()
{
//創建Producer,并配置發送消息所必需的信息。
ONSFactoryProperty factoryInfo;
factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");//您在控制臺創建的Group ID。
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX"); //設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );// 在控制臺創建的Topic。
factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");//消息內容。
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "XXX");// AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "XXX" );// AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
//create producer;
Producer *pProducer = ONSFactory::getInstance()->createProducer(factoryInfo);
//在發送消息前,必須調用start方法來啟動Producer,只需調用一次即可。
pProducer->start();
Message msg(
//Message Topic
factoryInfo.getPublishTopics(),
//Message Tag,可理解為Gmail中的標簽,對消息進行再歸類,方便Consumer指定過濾條件在消息隊列RocketMQ版的服務器過濾。
"TagA",
//Message Body,不能為空,消息隊列RocketMQ版不做任何干預,需要Producer與Consumer協商好一致的序列化和反序列化方式。
factoryInfo.getMessageContent()
);
// 設置代表消息的業務關鍵屬性,請盡可能全局唯一。
// 以方便您在無法正常收到消息情況下,可通過控制臺查詢消息并補發。
// 注意:不設置也不會影響消息正常收發。
msg.setKey("ORDERID_100");
//發送消息,只要不拋出異常,就代表發送成功。
try
{
SendResultONS sendResult = pProducer->send(msg);
}
catch(ONSClientException & e)
{
//自定義處理exception的細節。
}
// 在應用退出前,必須銷毀Producer對象,否則會導致內存泄露等問題。
pProducer->shutdown();
return 0;
}
同時,您也可以在控制臺頁面,找到您創建的Topic,在其操作列,單擊更多,在下拉列表中,選擇快速體驗,按需通過控制臺或Docker快速體驗。
查看消息是否發送成功
消息發送后,您可以在云消息隊列 RocketMQ 版控制臺查看消息發送狀態,步驟如下:
- 在實例所在頁面的左側導航欄,單擊消息查詢。
- 在消息查詢頁面,按需選擇查詢條件,單擊查詢。
存儲時間表示云消息隊列 RocketMQ 版服務端存儲這條消息的時間。如果查詢到此信息,表示消息已經成功發送到服務端。
調用TCP協議的SDK訂閱普通消息
消息發送成功后,需要啟動消費者來訂閱普通消息。請按需運行對應語言的示例代碼來啟動消費者,并測試訂閱消息的功能。請按照說明正確設置相關參數。
Java
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;
public class ConsumerTest {
public static void main(String[] args) {
Properties properties = new Properties();
// 您在控制臺創建的Group ID。
properties.put(PropertyKeyConst.GROUP_ID, "XXX");
// AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
properties.put(PropertyKeyConst.AccessKey, "XXX");
// AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
properties.put(PropertyKeyConst.SecretKey, "XXX");
// 設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
properties.put(PropertyKeyConst.NAMESRV_ADDR,
"XXX");
// 集群訂閱方式(默認)。
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
// 廣播訂閱方式。
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() { //訂閱多個Tag
public Action consume(Message message, ConsumeContext context) {
System.out.println("Receive: " + message);
return Action.CommitMessage;
}
});
//訂閱另外一個Topic,如需取消訂閱該Topic,請刪除該部分的訂閱代碼,重新啟動消費端即可。
consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() { //訂閱全部Tag。
public Action consume(Message message, ConsumeContext context) {
System.out.println("Receive: " + message);
return Action.CommitMessage;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}
.NET
using System;
using System.Threading;
using System.Text;
using ons;
// 從Broker拉取消息時要執行的回調函數。
public class MyMsgListener : MessageListener
{
public MyMsgListener()
{
}
~MyMsgListener()
{
}
public override ons.Action consume(Message value, ConsumeContext context)
{
Byte[] text = Encoding.Default.GetBytes(value.getBody());
Console.WriteLine(Encoding.UTF8.GetString(text));
return ons.Action.CommitMessage;
}
}
public class ConsumerExampleForEx
{
public ConsumerExampleForEx()
{
}
static void Main(string[] args) {
// 配置您的賬號,以下設置均可從控制臺獲取。
ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
// AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
// AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
// 您在控制臺創建的Group ID
factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "GID_example");
// 您在控制臺創建的Topic。
factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
// 設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
// 設置日志路徑。
factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");
// 集群消費。
// factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.CLUSTERING);
// 廣播消費。
// factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.BROADCASTING);
// 創建消費者實例。
PushConsumer consumer = ONSFactory.getInstance().createPushConsumer(factoryInfo);
// 訂閱Topics。
consumer.subscribe(factoryInfo.getPublishTopics(), "*", new MyMsgListener());
// 啟動客戶端實例。
consumer.start();
//該設置僅供Demo使用,實際生產中請保證進程不退出。
Thread.Sleep(300000);
// 在進程即將退出時,關閉消費者實例。
consumer.shutdown();
}
}
C/C++
#include "ONSFactory.h"
using namespace ons;
// MyMsgListener:創建消費消息的實例。
//pushConsumer拉取到消息后,會主動調用該實例的consumer函數。
class MyMsgListener : public MessageListener
{
public:
MyMsgListener()
{
}
virtual ~MyMsgListener()
{
}
virtual Action consume(Message &message, ConsumeContext &context)
{
//自定義消息處理細節。
return CommitMessage; //CONSUME_SUCCESS;
}
};
int main(int argc, char* argv[])
{
//pushConsumer創建和工作需要的參數,必須輸入。
ONSFactoryProperty factoryInfo;
factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "XXX");//您在控制臺創建的Group ID。
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX"); //設置TCP協議接入點,進入控制臺的實例詳情頁面的TCP協議客戶端接入點區域查看。
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );//您在控制臺創建的Topic。
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "XXX");//AccessKey ID,阿里云身份驗證標識。獲取方式,請參見本文前提條件中的獲取AccessKey。
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "XXX");//AccessKey Secret,阿里云身份驗證密鑰。獲取方式,請參見本文前提條件中的獲取AccessKey。
// 集群訂閱方式(默認)。
// factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty::CLUSTERING);
// 廣播訂閱方式。
// factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty::BROADCASTING);
//create pushConsumer
PushConsumer* pushConsumer = ONSFactory::getInstance()->createPushConsumer(factoryInfo);
//指定pushConsumer訂閱的消息Topic和Tag,注冊消息回調函數。
MyMsgListener msglistener;
pushConsumer->subscribe(factoryInfo.getPublishTopics(), "*",&msglistener );
//start pushConsumer
pushConsumer->start();
//注意:直到不再接收消息,才能調用shutdown;調用shutdown之后,Consumer退出,不能接收到任何消息。
//銷毀pushConsumer,在應用退出前,必須銷毀Consumer對象,否則會導致內存泄露等問題。
pushConsumer->shutdown();
return 0;
}
查看消息是否訂閱成功
- 在實例所在頁面的左側導航欄,單擊Group 管理。
- 在Group 管理頁面,單擊TCP 協議頁簽。
- 找到要查看的Group ID,在其操作列,單擊詳情。消費者狀態顯示為在線,且訂閱關系一致,說明訂閱成功。
更多信息
如需使用社區版TCP協議的SDK,請參見社區版TCP協議SDK。
文檔內容是否對您有幫助?