ApsaraMQ for RocketMQ:Envoyer et recevoir des messages transactionnels

Processus d'interaction

Le schéma suivant illustre le processus d'interaction des messages transactionnels.

Pour plus d'informations, consultez la rubrique Messages transactionnels.

Prérequis

• La bibliothèque dynamique C++ est installée. Pour plus d'informations, consultez la rubrique Installer la bibliothèque dynamique C++.

• Une paire AccessKey a été créée pour votre compte Alibaba Cloud. Pour plus d'informations, consultez la rubrique Créer une paire AccessKey.

Envoyer des messages transactionnels

1. Copiez le code suivant dans le fichier TransProducerDemo.cpp. Modifiez les paramètres correspondants, exécutez la commande g++ pour compiler le code, puis générez un fichier exécutable.

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

2. L'exemple de code suivant montre comment envoyer des messages transactionnels à l'aide du SDK client TCP pour C++ de l'édition Community :

   #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) {
           // Execute the local transaction. If the local transaction is executed, COMMIT_MESSAGE is returned. If the local transaction fails to be executed, ROLLBACK_MESSAGE is returned. If the execution status of the local transaction is unknown, UNKNOWN is returned. 
           // If UNKNOWN is returned, the scheduled task to query the status of the local transaction is triggered. 
           std::cout << "Execute Local Transaction,Received Message Topic:" << msg.getTopic() << ", MsgId:"
                     << msg.getBody() << std::endl;
           return UNKNOWN;
       }
   
       LocalTransactionState checkLocalTransaction(const MQMessageExt &msg) {
           // Query the execution status of the local transaction. If the local transaction is executed, COMMIT_MESSAGE is returned. If the local transaction fails to be executed, ROLLBACK_MESSAGE is returned. If the execution status of the local transaction is unknown, UNKNOWN is returned. 
           // If UNKNOWN is returned, wait until the next scheduled task to query the status of the local transaction is triggered. 
           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;
       // The ID of the group for which you applied in the ApsaraMQ for RocketMQ console. 
       TransactionMQProducer producer("GID_XXXXXXXXXXXXXXXX");
       // The TCP endpoint that you obtained from the Instance Details page in the ApsaraMQ for RocketMQ console. 
       producer.setNamesrvAddr("http://MQ_XXXXXXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
       // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
   	  // ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account, which are used for identity verification. 
       // The user channel. Default value: ALIYUN. 
       producer.setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
   
       // The transaction listener. 
       ExampleTransactionListener *exampleTransactionListener = new ExampleTransactionListener();
       producer.setTransactionListener(exampleTransactionListener);
       // After you configure the required parameters, start the producer. 
       producer.start();
       auto start = std::chrono::system_clock::now();
       int count = 3;
       for (int i = 0; i < count; ++i) {
           // The topic for which you applied in the ApsaraMQ for RocketMQ console. 
           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;
   }

Consommer des messages transactionnels

1. Copiez le code suivant dans le fichier ConsumerDemo.cpp. Modifiez les paramètres correspondants, exécutez la commande g++ pour compiler le code, puis générez un fichier exécutable.

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

2. L'exemple de code suivant montre comment consommer des messages transactionnels à l'aide du SDK client TCP pour C++ de l'édition Community :

   #include <iostream>\n#include <thread>\n#include "DefaultMQPushConsumer.h"\n\nusing namespace rocketmq;\n\n\nclass ExampleMessageListener : public MessageListenerConcurrently {\npublic:\n    ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {\n        for (auto item = msgs.begin(); item != msgs.end(); item++) {\n            std::cout << "Received Message Topic:" << item->getTopic() << ", MsgId:" << item->getMsgId() << std::endl;\n        }\n        return CONSUME_SUCCESS;\n    }\n};\n\nint main(int argc, char *argv[]) {\n    std::cout << "=======Before consuming messages=======" << std::endl;\n    // The ID of the group for which you applied in the ApsaraMQ for RocketMQ console. \n    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXXXXXXX");\n    // The TCP endpoint that you obtained from the Instance Details page in the ApsaraMQ for RocketMQ console. \n    consumer->setNamesrvAddr("http://MQ_INST_XXXXXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");\n    // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. \n\t  // ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account, which are used for identity verification. \n    // The user channel. Default value: ALIYUN. \n    consumer->setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");\n    auto start = std::chrono::system_clock::now();\n\n\n\n    // Use a custom listener function to process the received message and return the result. \n    ExampleMessageListener *messageListener = new ExampleMessageListener();\n    consumer->subscribe("YOURTOPIC", "*");\n    consumer->registerMessageListener(messageListener);\n\n    // The preparation is complete. You must invoke the startup function to start the consumer. \n    // ********************************************\n    // 1. Before you start the consumer, make sure that the subscription is configured. \n    // 2. Make sure that the subscriptions of consumers in the same group are consistent. \n    // *********************************************\n    consumer->start();\n\n    // Keep the main thread running until the process is shut down. \n    std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));\n    consumer->shutdown();\n    std::cout << "=======After consuming messages======" << std::endl;\n    return 0;\n}