全部產品
Search
文件中心

ApsaraMQ for RocketMQ:收發普通訊息

更新時間:Jul 01, 2024

普通訊息是指阿里雲雲訊息佇列 RocketMQ 版中無特性的訊息,區別於有特性的定時訊息、順序訊息和事務訊息。本文提供使用TCP協議下的社區版C++ SDK收發普通訊息的範例程式碼供您參考。

前提條件

發送普通訊息

  1. 將以下代碼拷貝到ProducerDemo.cpp檔案中,修改相應的參數後,使用g++命令進行編譯,產生可執行檔,即可直接運行。

    g++ -o producer_demo -std=c++11 -lz -lrocketmq ProducerDemo.cpp

  2. 發送普通訊息的範例程式碼如下。

    #include <iostream>
    #include <chrono>
    #include <thread>
    #include "DefaultMQProducer.h"
    
    using namespace std;
    using namespace rocketmq;
    
    int main() {
        std::cout << "=======Before sending messages=======" << std::endl;
        //您在阿里雲訊息佇列RocketMQ控制台上申請的GID。
        DefaultMQProducer producer("GID_XXXXXXXXX");
        //設定TCP協議存取點,從阿里雲RocketMQ控制台的執行個體詳情頁面擷取。
        producer.setNamesrvAddr("http://MQ_INST_XXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
        //請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
    	  //ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET分別為阿里雲帳號的AccessKey ID和AccessKey Secret,用於身分識別驗證。
        //使用者渠道,預設值為:ALIYUN。
        producer.setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
    
        //請確保參數設定完成之後啟動Producer。
        producer.start();
        auto start = std::chrono::system_clock::now();
        int count = 32;
        for (int i = 0; i < count; ++i) {
            //發送訊息時請設定您在阿里雲訊息佇列RocketMQ控制台上申請的Topic。
            MQMessage msg("YOURTOPIC","HiTAG","HelloCPPSDK.");
            try {
                SendResult sendResult = producer.send(msg);
                std::cout <<"SendResult:"<<sendResult.getSendStatus()<< ", Message ID: " << sendResult.getMsgId() << std::endl;
                this_thread::sleep_for(chrono::seconds(1));
            } catch (MQException e) {
                std::cout << "ErrorCode: " << e.GetError() << " Exception:" << e.what() << std::endl;
            }
        }
        auto interval = std::chrono::system_clock::now() - start;
        std::cout << "Send " << count << " messages OK, costs "
                  << std::chrono::duration_cast<std::chrono::milliseconds>(interval).count() << "ms" << std::endl;
    
        producer.shutdown();
        std::cout << "=======After sending messages=======" << std::endl;
        return 0;
    }

消費普通訊息

  1. 將以下代碼拷貝到ConsumerDemo.cpp檔案中,修改相應的參數後,使用g++命令進行編譯,產生可執行檔,即可直接運行。

    g++ -o consumer_demo -std=c++11 -lz -lrocketmq ConsumerDemo.cpp

  2. 消費普通訊息的範例程式碼如下。

    #include <iostream>
    #include <thread>
    #include "DefaultMQPushConsumer.h"
    
    using namespace rocketmq;
    
    
    class ExampleMessageListener : public MessageListenerConcurrently {
    public:
        ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {
            for (auto item = msgs.begin(); item != msgs.end(); item++) {
                std::cout << "Received Message Topic:" << item->getTopic() << ", MsgId:" << item->getMsgId() << std::endl;
            }
            return CONSUME_SUCCESS;
        }
    };
    
    int main(int argc, char *argv[]) {
        std::cout << "=======Before consuming messages=======" << std::endl;
        //您在阿里雲訊息佇列RocketMQ控制台上申請的GID。
        DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXXXXXXX");
        //設定TCP協議存取點,從阿里雲訊息佇列RocketMQ控制台的執行個體詳情頁面擷取。
        consumer->setNamesrvAddr("http://MQ_INST_XXXXXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
        //請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
    	  //ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET分別為阿里雲帳號的AccessKey ID和AccessKey Secret,用於身分識別驗證。
        //使用者渠道,預設值為:ALIYUN。
        consumer->setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
        auto start = std::chrono::system_clock::now();
    
    
    
        //請註冊自訂偵聽函數用來處理接收到的訊息,並返迴響應的處理結果。
        ExampleMessageListener *messageListener = new ExampleMessageListener();
        consumer->subscribe("YOURTOPIC", "*");
        consumer->registerMessageListener(messageListener);
    
        //準備工作完成,必須調用啟動函數,才可以正常工作。
        // ********************************************
        // 1.確保訂閱關係的設定在啟動之前完成。
        // 2.確保相同GID下面的消費者的訂閱關係一致。
        // *********************************************
        consumer->start();
    
        // 保持主線程運行直到進程結束。
        std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
        consumer->shutdown();
        std::cout << "=======After consuming messages======" << std::endl;
        return 0;
    }