ApsaraMQ for RocketMQ:Enviar e receber mensagens transacionais

Siga as etapas abaixo para enviar uma mensagem transacional:

1. Envie uma half message e execute uma transação local. Exemplo de código:

   #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. O exemplo de código a seguir demonstra como confirmar o status de uma mensagem transacional:

    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;
        }
    }               

Mecanismo de verificação de status da transação

• Por que implementar o mecanismo de verificação de status da transação ao enviar mensagens transacionais?

  Se uma half message for enviada, mas TransactionStatus.Unknow for retornado ou nenhum status for confirmado para a transação local devido ao encerramento da aplicação, o status da half message permanecerá desconhecido para o broker do ApsaraMQ for RocketMQ. Por isso, o broker envia periodicamente uma solicitação a um produtor no cluster de produtores para verificar o status dessa half message. Ao receber a solicitação de verificação, o produtor verifica e confirma o status final da transação local correspondente à half message.

• Qual é a função da lógica de negócio quando o método check é chamado?

  O método check para mensagens transacionais deve conter a lógica de verificação da consistência da transação. Após o envio de uma mensagem transacional, o ApsaraMQ for RocketMQ precisa chamar a operação de API LocalTransactionCheckerg para responder à solicitação do broker sobre o status da transação local. Portanto, o método de verificação de mensagens transacionais deve cumprir os seguintes objetivos:

  1. Verificar o status (confirmado ou revertido) da transação local correspondente à half message.

  2. Confirmar ao broker o status da transação local correspondente à half message.