ApsaraMQ for RocketMQ:İşlemsel mesajları gönderme ve alma

Bir işlemsel mesajı göndermek için şu adımları uygulayın:

1. Yarı mesaj gönderin ve yerel bir işlemi yürütün. Örnek kod:

   #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. Aşağıdaki örnek kod, bir işlemsel mesajın durumunun nasıl onaylanacağını gösterir:

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

İşlem durumu denetimi mekanizması

• İşlemsel mesajlar gönderilirken neden işlem durumu denetimi mekanizmasının uygulanması gerekir?

  Yarı mesaj gönderilir ancak TransactionStatus.Unknow döndürülürse veya uygulama çıkışı nedeniyle yerel işlem için hiçbir durum onaylanmazsa yarı mesajın durumu ApsaraMQ for RocketMQ broker tarafından bilinmez. Bu nedenle broker, periyodik olarak üretici kümesindeki bir üreticiye yarı mesajın durumunu kontrol etmek için istek gönderir. Durum denetimi isteğini alan üretici, yarı mesajla ilişkili yerel işlemin nihai durumunu denetler ve onaylar.

• Denetim yöntemi geri çağrıldığında iş mantığı ne yapar?

  İşlemsel mesajlar için denetim yöntemi, işlem tutarlılığını kontrol etmek için kullanılan mantığı içermelidir. Bir işlemsel mesaj gönderildikten sonra ApsaraMQ for RocketMQ, yerel işlem durumu için broker'dan gelen isteğe yanıt vermek üzere LocalTransactionCheckerg API işlemini çağırmalıdır. Bu nedenle işlemsel mesajları denetlemek için kullanılan yöntemin aşağıdaki hedeflere ulaşması gerekir:

  1. Yarı mesajla ilişkili yerel işlemin durumunu (onaylandı veya geri alındı) denetleyin.

  2. Yarı mesajla ilişkili yerel işlemin durumunu broker'a onaylayın.