ApsaraMQ for RocketMQ：收發事務訊息

互動流程

事務訊息互動流程如下圖所示。

更多資訊，請參見事務訊息。

前提條件

• 安裝CPP動態庫

• 擷取阿里雲存取金鑰AccessKey ID和AccessKey Secret。更多資訊，請參見建立AccessKey。

發送事務訊息

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

   g++ -o trans_producer_demo -std=c++11 -lz -lrocketmq TransProducerDemo.cpp

2. 發送事務訊息的範例程式碼如下。

   #include <iostream>
   #include <chrono>
   #include <thread>
   #include "TransactionMQProducer.h"
   #include "MQClientException.h"
   #include "TransactionListener.h"
   
   using namespace std;
   using namespace rocketmq;
   
   class ExampleTransactionListener : public TransactionListener {
   public:
       LocalTransactionState executeLocalTransaction(const MQMessage &msg, void *arg) {
           //執行本地事務，成功返回COMMIT_MESSAGE，失敗返回ROLLBACK_MESSAGE，不確定返回UNKNOWN。
           //如果返回UNKNOWN，則會觸發定時任務回查狀態。
           std::cout << "Execute Local Transaction,Received Message Topic:" << msg.getTopic() << ", MsgId:"
                     << msg.getBody() << std::endl;
           return UNKNOWN;
       }
   
       LocalTransactionState checkLocalTransaction(const MQMessageExt &msg) {
           //回查本地事務執行情況，成功返回COMMIT_MESSAGE，失敗返回ROLLBACK_MESSAGE，不確定返回UNKNOWN。
           //如果返回UNKNOWN，則等待下次定時任務回查。
           std::cout << "Check Local Transaction,Received Message Topic:" << msg.getTopic() << ", MsgId:" << msg.getMsgId()
                     << std::endl;
           return COMMIT_MESSAGE;
       }
   };
   
   int main() {
       std::cout << "=======Before sending messages=======" << std::endl;
       //您在阿里雲訊息佇列RocketMQ版控制台上申請的GID。
       TransactionMQProducer producer("GID_XXXXXXXXXXXXXXXX");
       //設定TCP協議存取點，從阿里雲訊息佇列RocketMQ版控制台的執行個體詳情頁面擷取。
       producer.setNamesrvAddr("http://MQ_XXXXXXXXXXXX.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");
   
       //本地事務執行和回查函數。
       ExampleTransactionListener *exampleTransactionListener = new ExampleTransactionListener();
       producer.setTransactionListener(exampleTransactionListener);
       //請確保參數設定完成之後啟動Producer。
       producer.start();
       auto start = std::chrono::system_clock::now();
       int count = 3;
       for (int i = 0; i < count; ++i) {
           //發送訊息時請設定您在阿里雲訊息佇列RocketMQ版控制台上申請的Topic。
           MQMessage msg("YOUR TRANSACTION TOPIC", "HiTAG", "Hello,CPP SDK, Transaction Message.");
           try {
               SendResult sendResult = producer.sendMessageInTransaction(msg, &i);
               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;
   
       std::cout << "Wait for local transaction check..... " << std::endl;
       for (int i = 0; i < 6; ++i) {
           this_thread::sleep_for(chrono::seconds(10));
           std::cout << "Running "<< i*10 + 10 << " Seconds......"<< 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;
   }