ApsaraMQ for RocketMQ:Envoyer et recevoir des messages transactionnels

Suivez les étapes ci-dessous pour envoyer un message transactionnel :

1. Envoyez un demi-message et exécutez une transaction locale. Exemple de code :

   #include "ONSFactory.h"
   #include "ONSClientException.h"
   using namespace ons;
   
       class MyLocalTransactionExecuter : LocalTransactionExecuter
       {
           MyLocalTransactionExecuter()
           {
           }
   
           ~MyLocalTransactionExecuter()
           {
           }
           virtual TransactionStatus execute(Message &value)
           {
                   // The message ID. Two messages can have the same message body but not the same ID. You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console. )
                   string msgId = value.getMsgID();
                   // Calculate the message body by using an algorithm such as CRC32 and MD5. 
                   // The message ID and CRC32 ID are used to prevent duplicate messages. 
                   // You do not need to specify the message ID or CRC32 ID if your business is idempotent. Otherwise, specify the message ID or CRC32 ID to ensure idempotence. 
                   // To prevent duplicate messages, we recommend that you calculate the message body by using the CRC32 or MD5 algorithm. 
                   TransactionStatus transactionStatus = Unknow;
                   try {
                       boolean isCommit = The execution result of the local transaction;
                       if (isCommit) {
                           // Commit the message if the local transaction is executed. 
                           transactionStatus = CommitTransaction;
                       } else {
                           // Roll back the message if the local transaction failed to be executed. 
                           transactionStatus = RollbackTransaction;
                       }
                   } catch (...) {
                       //exception handle
                   }
                   return transactionStatus;
           }
       }
   
       int main(int argc, char* argv[])
       {
           // The parameters that are required to create the producer and send messages. 
           ONSFactoryProperty factoryInfo;
           // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
           // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX");
           // The topic that you created in the ApsaraMQ for RocketMQ console. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
           // The message content. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
           // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
           // The AccessKey ID that is used for authentication. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
   		    // The AccessKey secret that is used for authentication. 
           factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
   
           // Create the producer. ApsaraMQ for RocketMQ does not release pChecker. You must release pChecker by yourself. 
           MyLocalTransactionChecker *pChecker = new MyLocalTransactionChecker();
           g_producer = ONSFactory::getInstance()->createTransactionProducer(factoryInfo,pChecker);
   
           // Before you send the message, call the start() method only once to start the producer. 
           pProducer->start();
   
           Message msg(
               //Message Topic
               factoryInfo.getPublishTopics(),
               // The message tag. A message tag is similar to a Gmail tag and is used by consumers to sort and filter messages in the ApsaraMQ for RocketMQ broker.        
               "TagA",
               // The message body. You cannot leave this parameter empty. ApsaraMQ for RocketMQ does not process message bodies. The producer and consumer must agree on the methods that are used to serialize and deserialize message bodies. 
               factoryInfo.getMessageContent()
           );
   
           // The message key. The key is the business-specific attribute of a message and must be globally unique whenever possible. 
           // If you cannot receive a message as expected, you can use the key to query the message in the ApsaraMQ for RocketMQ console. 
           // Note: You can send and receive a message even if you do not specify the key. 
           msg.setKey("ORDERID_100");
   
           // Send the message. If no exception is thrown, the message is sent.    
           try
           {
               // ApsaraMQ for RocketMQ does not release pExecuter. You must release pExecuter by yourself. 
               MyLocalTransactionExecuter pExecuter = new MyLocalTransactionExecuter();
               SendResultONS sendResult = pProducer->send(msg,pExecuter);
           }
           catch(ONSClientException & e)
           {
               // Specify the logic to handle exceptions. 
           }
           // Before you exit the application, destroy the producer. Otherwise, issues such as memory leaks occur. 
           pProducer->shutdown();
   
           return 0;
   
       }        

2. L'exemple de code suivant montre comment valider le statut d'un message transactionnel :

    class MyLocalTransactionChecker : LocalTransactionChecker
    {
        MyLocalTransactionChecker()
        {
        }
   
        ~MyLocalTransactionChecker()
        {
        }
   
        virtual TransactionStatus check(Message &value)
        {
            // The message ID. Two messages can have the same message body but not the same ID. You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console.
            string msgId = value.getMsgID();
            // Calculate the message body by using an algorithm such as CRC32 and MD5. 
            // The message ID and CRC32 ID are used to prevent duplicate messages. 
            // You do not need to specify the message ID or CRC32 ID if your business is idempotent. Otherwise, specify the message ID or CRC32 ID to ensure idempotence. 
            // To prevent duplicate messages, we recommend that you calculate the message body by using the CRC32 or MD5 algorithm. 
            TransactionStatus transactionStatus = Unknow;
            try {
                boolean isCommit = The execution result of the local transaction;
                if (isCommit) {
                    // Commit the message if the local transaction is executed. 
                    transactionStatus = CommitTransaction;
                } else {
                    // Roll back the message if the local transaction failed to be executed. 
                    transactionStatus = RollbackTransaction;
                }
            } catch(...) {
                //exception error
            }
            return transactionStatus;
        }
    }               

Mécanisme de vérification du statut de transaction

• Pourquoi est-il nécessaire de mettre en œuvre un mécanisme de vérification du statut de transaction lors de l'envoi de messages transactionnels ?

  Si un demi-message est envoyé mais que TransactionStatus.Unknow est renvoyé, ou si aucun statut n'est validé pour la transaction locale en raison de l'arrêt de l'application, le courtier ApsaraMQ for RocketMQ ignore le statut du demi-message. Par conséquent, le courtier envoie périodiquement une requête à un producteur du cluster de producteurs pour vérifier le statut d'un demi-message. Après réception de la requête de vérification, le producteur vérifie et valide le statut final de la transaction locale correspondant au demi-message.

• Quelle logique métier doit être appliquée lors du rappel de la méthode de vérification ?

  La méthode de vérification des messages transactionnels doit inclure la logique permettant de vérifier la cohérence transactionnelle. Après l'envoi d'un message transactionnel, ApsaraMQ for RocketMQ doit appeler l'opération API LocalTransactionCheckerg pour répondre à la requête du courtier concernant le statut de la transaction locale. Par conséquent, la méthode utilisée pour vérifier les messages transactionnels doit atteindre les objectifs suivants :

  1. Vérifier le statut (validé ou annulé) de la transaction locale correspondant au demi-message.

  2. Valider auprès du courtier le statut de la transaction locale correspondant au demi-message.