Send and subscribe to transactional messages

package com.alibaba.webx.TryHsf.app1;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class TransactionProducerClient {
 private final static Logger log = ClientLogger.getLog(); // Configure logging to facilitate troubleshooting. 

 public static void main(String[] args) throws InterruptedException {
     final BusinessService businessService = new BusinessService(); // Your local business service. 
     Properties properties = new Properties();
        // The ID of the group that you create in the Message Queue for Apache RocketMQ console. Note: Transactional messages cannot share group IDs with other types of messages. 
     properties.put(PropertyKeyConst.GROUP_ID,"XXX");
        // The AccessKey ID that you create in the RAM console for identity authentication. 
     properties.put(PropertyKeyConst.AccessKey,"XXX");
        // The AccessKey secret that you create in the RAM console for identity authentication. 
     properties.put(PropertyKeyConst.SecretKey,"XXX");
        // The TCP endpoint of your instance. To obtain the TCP endpoint, log on to the Message Queue for Apache RocketMQ console. In the left-side navigation pane, click Instances. On the Instances page, click the name of your instance. On the Instance Details page that appears, go to the Basic Information section and view the TCP endpoint on the Endpoints tab. 
     properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXX");

     TransactionProducer producer = ONSFactory.createTransactionProducer(properties,
             new LocalTransactionCheckerImpl());
     producer.start();
     Message msg = new Message("Topic","TagA","Hello MQ transaction===".getBytes());
     try {
             SendResult sendResult = producer.send(msg, new LocalTransactionExecuter() {
                 @Override
                 public TransactionStatus execute(Message msg, Object arg) {
                     // The ID of the message. Two messages may have the same message body but cannot have the same ID. The current message is a half message. Therefore, its message ID cannot be queried in the Message Queue for Apache RocketMQ console. 
                     String msgId = msg.getMsgID();
                     // Calculate the message body by using an algorithm such as CRC32 and MD5. 
                     long crc32Id = HashUtil.crc32Code(msg.getBody());
                     // 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 service itself achieves idempotence. Otherwise, specify the message ID or CRC32 ID to ensure idempotence. 
                     // To prevent duplicate messages, calculate the message body by using the CRC32 or MD5 algorithm. 
                     Object businessServiceArgs = new Object();
                     TransactionStatus transactionStatus = TransactionStatus.Unknow;
                     try {
                         boolean isCommit =
                             businessService.execbusinessService(businessServiceArgs);
                         if (isCommit) {
                             // Commit the message if the local transaction succeeds. 
                             transactionStatus = TransactionStatus.CommitTransaction;
                         } else {
                             // Roll back the message if the local transaction fails. 
                             transactionStatus = TransactionStatus.RollbackTransaction;
                         }
                     } catch (Exception e) {
                         log.error("Message Id:{}", msgId, e);
                     }
                     System.out.println(msg.getMsgID());
                     log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
                     return transactionStatus;
                 }
             }, null);
         }
         catch (Exception e) {
                // Specify the logic to resend or persist the message if the message fails to be sent and needs to be resent. 
             System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
             e.printStackTrace();
         }
     // Use the demo to prevent the process from exiting. This is not required in actual use. 
     TimeUnit.MILLISECONDS.sleep(Integer.MAX_VALUE);
 }
}                        

After the local transaction is executed, the Message Queue for Apache RocketMQ broker must be notified of the transaction status of the current message, no matter whether the execution is successful or fails. The Message Queue for Apache RocketMQ broker can be notified in one of the following ways:

• Commit the status after the local transaction is executed.
• Wait until the Message Queue for Apache RocketMQ broker sends a request to check the transaction status of the message.

A transaction can be in one of the following states:

• TransactionStatus.CommitTransaction: The transaction is committed. The consumer can consume the message.
• TransactionStatus.RollbackTransaction: The transaction is rolled back. The message is discarded and cannot be consumed.
• TransactionStatus.Unknow: The transaction is in an unknown state. The Message Queue for Apache RocketMQ broker is expected to query the status of the local transaction that corresponds to the message from the message producer.

public class LocalTransactionCheckerImpl implements LocalTransactionChecker {
   private final static Logger log = ClientLogger.getLog();
   final  BusinessService businessService = new BusinessService();

   @Override
   public TransactionStatus check(Message msg) {
       // The ID of the message. Two messages may have the same message body but cannot have the same ID. The current message is a half message. Therefore, its message ID cannot be queried in the Message Queue for Apache RocketMQ console. 
       String msgId = msg.getMsgID();
       // Calculate the message body by using an algorithm such as CRC32 and MD5. 
       long crc32Id = HashUtil.crc32Code(msg.getBody());
       // 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 service itself achieves idempotence. Otherwise, specify the message ID or CRC32 ID to ensure idempotence. 
       // To prevent duplicate messages, calculate the message body by using the CRC32 or MD5 algorithm. 
       // The parameter object of your business service. Specify the object based on your business service. 
       Object businessServiceArgs = new Object();
       TransactionStatus transactionStatus = TransactionStatus.Unknow;
       try {
           boolean isCommit = businessService.checkbusinessService(businessServiceArgs);
           if (isCommit) {
               // Commit the message if the local transaction succeeds. 
               transactionStatus = TransactionStatus.CommitTransaction;
           } else {
               // Roll back the message if the local transaction fails. 
               transactionStatus = TransactionStatus.RollbackTransaction;
           }
       } catch (Exception e) {
           log.error("Message Id:{}", msgId, e);
       }
       log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
       return transactionStatus;
   }
 }                        

Utility class

import java.util.zip.CRC32;
public class HashUtil {
    public static long crc32Code(byte[] bytes) {
        CRC32 crc32 = new CRC32();
        crc32.update(bytes);
        return crc32.getValue();
    }
}