全部產品
Search

搜尋本產品

    ApsaraMQ for RocketMQ：調用TCP協議的SDK收發普通訊息

    文件中心

    ApsaraMQ for RocketMQ：調用TCP協議的SDK收發普通訊息

    更新時間：Dec 27, 2024

    在控制台建立完所有資源之後，您可以調用雲訊息佇列 RocketMQ 版的TCP協議的SDK收發普通訊息。

    前提條件

    • 建立資源

      說明

      本文以普通訊息為例進行說明，因此您建立的普通訊息的Topic不能用來收發其他類型的訊息，包括定時、延時、順序和事務訊息。換言之，切勿混用不同訊息類型的Topic。

    • 建立AccessKey

    下載並安裝TCP協議的SDK

    說明

    商業版SDK和社區版SDK相比，提供了更加豐富的功能特性並具有更高的穩定性保障，推薦您使用阿里雲雲訊息佇列 RocketMQ 版商業版SDK進行接入。社區版SDK僅在開源RocketMQ遷移上雲且不希望修改代碼時使用。

    雲訊息佇列 RocketMQ 版提供了以下商業版TCP協議的SDK，請按需擷取相應語言的用戶端SDK。

    Java SDK

    C/C++ SDK

    .NET 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 版控制台查看訊息發送狀態，步驟如下：

    1. 在執行個體所在頁面的左側導覽列，單擊消息查询。
    2. 在消息查询頁面，按需選取查詢條件，單擊查询。

      存储时间表示雲訊息佇列 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;

}

    查看訊息是否訂閱成功

    1. 在執行個體所在頁面的左側導覽列，單擊Group 管理。
    2. 在Group 管理頁面，單擊TCP 协议頁簽。
    3. 找到要查看的Group ID，在其操作列，單擊详情。
      消费者状态顯示為在线，且訂閱關係一致，說明訂閱成功。