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.");
    }
}

Implémenter le vérificateur de transaction

Le courtier invoque le vérificateur de transaction lorsqu'il ne peut pas déterminer le statut d'un message semi-validé. Cela se produit dans les cas suivants :

• L'objet LocalTransactionExecuter renvoie la valeur TransactionStatus.Unknow.

• Le producteur s'arrête avant d'avoir validé le statut de la transaction.

Dans ces deux cas, le courtier envoie périodiquement une demande de vérification de statut à un producteur du cluster de producteurs. Le producteur vérifie alors la transaction locale et signale le statut 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;
    }
}

Votre méthode check doit :

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

2. Renvoyer le statut de la transaction au courtier : TransactionStatus.CommitTransaction, TransactionStatus.RollbackTransaction ou TransactionStatus.Unknow.