ApsaraMQ for RocketMQ:Send and receive transactional messages

package com.aliyun.openservices.tcp.example.producer;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
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.Date;
import java.util.Properties;

public class SimpleTransactionProducer {

    public static void main(String[] args) {

        Properties properties = new Properties();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        // The consumer group ID for transactional messages cannot be the same
        // as the ID used for other message types.
        properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // Get AccessKey credentials from environment variables
        // to avoid hardcoding sensitive information.
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint. You can find this value in the TCP Endpoint section
        // of the Instance Details page in the ApsaraMQ for RocketMQ console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");

        // Register a transaction checker before you create the producer.
        // The broker calls this checker to verify unresolved transactions.
        LocalTransactionCheckerImpl localTransactionChecker = new LocalTransactionCheckerImpl();
        TransactionProducer transactionProducer = ONSFactory.createTransactionProducer(properties, localTransactionChecker);
        transactionProducer.start();

        Message msg = new Message("<your-topic>", "TagA", "Hello MQ transaction===".getBytes());

        for (int i = 0; i < 3; i++) {
            try {
                SendResult sendResult = transactionProducer.send(msg, new LocalTransactionExecuter() {
                    @Override
                    public TransactionStatus execute(Message msg, Object arg) {
                        // Run your local transaction logic here.
                        System.out.println("Execute the local transaction and commit the transaction status.");
                        return TransactionStatus.CommitTransaction;
                    }
                }, null);
                assert sendResult != null;
            } catch (ONSClientException e) {
                // Handle the failure: retry sending or persist the message for later processing.
                System.out.println(new Date() + " Send mq message failed! Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        System.out.println("Send transaction message success.");
    }
}

Implementar o verificador de transações

O broker invoca o verificador de transações quando não consegue determinar o status de uma half message. Isso ocorre nas seguintes situações:

• O LocalTransactionExecuter retorna TransactionStatus.Unknow.

• O produtor encerra a execução antes de confirmar o status da transação.

Em ambos os casos, o broker envia periodicamente uma solicitação de verificação de status a um produtor no cluster de produtores. O produtor verifica a transação local e reporta o status final.

// Transaction checker implementation
class LocalTransactionCheckerImpl implements LocalTransactionChecker {

    @Override
    public TransactionStatus check(Message msg) {
        System.out.println("Received transaction status check request. MsgId: " + msg.getMsgID());
        // Query the local transaction status and return the result.
        // Return CommitTransaction, RollbackTransaction, or Unknow.
        return TransactionStatus.CommitTransaction;
    }
}

O método check deve:

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

2. Retornar o status da transação ao broker: TransactionStatus.CommitTransaction, TransactionStatus.RollbackTransaction ou TransactionStatus.Unknow.