Envío y recepción de mensajes transaccionales - ApsaraMQ for RocketMQ

Los mensajes transaccionales garantizan la atomicidad entre una transacción local y la entrega posterior del mensaje. Si la transacción local se confirma, el mensaje se entrega a los consumidores; si se revierte, el mensaje se descarta. Esto evita dos modos de fallo: entregar un mensaje correspondiente a una transacción que nunca se completó o completar una transacción sin entregar su mensaje asociado.

A continuación, los ejemplos utilizan el SDK de cliente TCP para Java (Community Edition) para implementar un productor y un consumidor de mensajes transaccionales.

Funcionamiento de los mensajes transaccionales

Un mensaje transaccional pasa por cinco etapas entre el productor y el broker de ApsaraMQ for RocketMQ:

El productor envía un mensaje parcial al broker. Este mensaje se almacena en el broker, pero aún no se entrega a los consumidores.
El broker persiste el mensaje parcial y devuelve un acuse de recibo (ACK) al productor.
El productor ejecuta la transacción local; por ejemplo, insertar un pedido en una base de datos.
El productor confirma el resultado de la transacción al broker. Según el resultado de la transacción local, el productor envía uno de tres estados:
- LocalTransactionState.COMMIT_MESSAGE: La transacción local se completó correctamente. El broker entrega el mensaje a los consumidores.
- LocalTransactionState.ROLLBACK_MESSAGE: La transacción local falló. El broker descarta el mensaje.
- LocalTransactionState.UNKNOW: El resultado es incierto. El broker comprueba el estado posteriormente.
El broker inicia una verificación del estado de la transacción si no recibe una confirmación o reversión definitiva. Esto ocurre cuando el productor devuelve UNKNOW, falla antes de confirmar un estado o pierde la conectividad de red. El broker llama periódicamente al listener de verificación de transacciones del productor hasta recibir una confirmación o una reversión.

Interaction process between the producer, broker, and transaction check listener

Para obtener más información sobre el modelo de mensajes transaccionales, consulte Transactional messages.

Requisitos previos

Antes de empezar, asegúrese de haber cumplido con lo siguiente:

Descargado la Community Edition del SDK para Java 4.5.2 o posterior desde la página de descargas de RocketMQ
Prepared the environment
Created an AccessKey pair para su cuenta de Alibaba Cloud

Envío de mensajes transaccionales

Para enviar un mensaje transaccional se requieren dos componentes:

Un productor que envíe el mensaje parcial y ejecute la transacción local.
Un listener de verificación de transacciones que responda a las solicitudes de comprobación de estado provenientes del broker.

Paso 1: Implementar el listener de verificación de transacciones

El broker invoca este listener cuando necesita verificar el resultado de una transacción local. Después de enviar un mensaje transaccional, ApsaraMQ for RocketMQ llama a la operación de API LocalTransactionChecker para responder a la solicitud de comprobación de estado del broker. Su implementación debe consultar el estado real de la transacción (por ejemplo, comprobar si existe un pedido en la base de datos) y devolver el estado correspondiente.

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;

public class LocalTransactionCheckerImpl implements TransactionCheckListener {

    @Override
    public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
        System.out.println("Received transaction status check request. MsgId: " + msg.getMsgId());

        // Query your business data to determine the transaction outcome.
        // Example: Check whether the order associated with this message
        // was committed to the database.
        //
        // boolean orderExists = orderService.existsByOrderId(msg.getKeys());
        // if (orderExists) {
        //     return LocalTransactionState.COMMIT_MESSAGE;
        // }
        // return LocalTransactionState.ROLLBACK_MESSAGE;

        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

Paso 2: Enviar un mensaje parcial y ejecutar la transacción local

El productor envía un mensaje parcial al broker y ejecuta inmediatamente la transacción local en el callback executeLocalTransactionBranch. Según el valor devuelto, el broker confirma, revierte o programa una verificación de estado para el mensaje.

Reemplace los siguientes marcadores de posición con sus valores reales:

Marcador de posición	Descripción	Ejemplo
`YOUR TRANSACTION GROUP ID`	El ID de grupo creado en la consola de ApsaraMQ for RocketMQ	GID_tx_order
`YOUR TRANSACTION TOPIC`	El topic creado en la consola de ApsaraMQ for RocketMQ	tx_order_topic
`YOUR MESSAGE TAG`	Una etiqueta para filtrar mensajes	TagA
`http://xxxx.mq-internet.aliyuncs.com:80`	El endpoint obtenido de la consola de ApsaraMQ for RocketMQ, con el formato `http://MQ_INST_XXXX.aliyuncs.com:80`	`http://MQ_INST_1234.mq-internet.aliyuncs.com:80`

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQTransactionProducer {

    // Load AccessKey credentials from environment variables.
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a transactional producer with message trace enabled.
        // To disable message trace, use:
        // new TransactionMQProducer("YOUR TRANSACTION GROUP ID", getAclRPCHook());
        TransactionMQProducer producer = new TransactionMQProducer(
            null, "YOUR TRANSACTION GROUP ID", getAclRPCHook(), true, null);

        // Set the endpoint from the ApsaraMQ for RocketMQ console.
        producer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");

        // Required for message trace on Alibaba Cloud.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Register the transaction check listener (see Step 1).
        producer.setTransactionCheckListener(new LocalTransactionCheckerImpl());

        producer.start();

        for (int i = 0; i < 10; i++) {
            try {
                Message message = new Message(
                    "YOUR TRANSACTION TOPIC",
                    "YOUR MESSAGE TAG",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Send the half message and execute the local transaction.
                SendResult sendResult = producer.sendMessageInTransaction(
                    message,
                    new LocalTransactionExecuter() {
                        @Override
                        public LocalTransactionState executeLocalTransactionBranch(
                                Message msg, Object arg) {
                            System.out.println("Executing local transaction: " + msg);
                            // Run your business logic here (e.g., insert order into DB).
                            // Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW
                            // based on the outcome.
                            return LocalTransactionState.UNKNOW;
                        }
                    },
                    null);
                assert sendResult != null;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Estados de la transacción

Estado	Significado	Comportamiento del broker
`LocalTransactionState.COMMIT_MESSAGE`	La transacción local se completó correctamente	Entrega el mensaje a los consumidores
`LocalTransactionState.ROLLBACK_MESSAGE`	La transacción local falló	Descarta el mensaje
`LocalTransactionState.UNKNOW`	El resultado aún no se ha determinado	Verifica periódicamente el estado de la transacción mediante el listener de comprobación

Importancia de la verificación del estado de la transacción

Si el productor devuelve UNKNOW, falla o pierde la conectividad después de enviar el mensaje parcial, el broker no puede determinar si debe entregar o descartar el mensaje. La verificación del estado de la transacción resuelve esta situación al consultar periódicamente al productor hasta recibir una confirmación o reversión definitiva.

Su implementación de LocalTransactionCheckerImpl debe:

Consultar el estado de la transacción local correspondiente al mensaje parcial; por ejemplo, buscar el pedido en la base de datos.
Devolver el estado correcto de LocalTransactionState al broker.

Suscripción a mensajes transaccionales

Suscríbase a los mensajes transaccionales de la misma manera que a los mensajes normales. El consumidor no tiene conocimiento de la transacción, ya que recibe el mensaje únicamente después de que el broker lo confirma.

Reemplace los siguientes marcadores de posición con sus valores reales:

Marcador de posición	Descripción	Ejemplo
`YOUR GROUP ID`	El ID del grupo de consumidores creado en la consola de ApsaraMQ for RocketMQ	GID_consumer_order
`YOUR TOPIC`	El topic al que desea suscribirse	tx_order_topic
`http://xxxx.mq-internet.aliyuncs.com:80`	El endpoint obtenido de la consola de ApsaraMQ for RocketMQ	`http://MQ_INST_1234.mq-internet.aliyuncs.com:80`

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

public class RocketMQPushConsumer {

    // Load AccessKey credentials from environment variables.
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a push consumer with message trace enabled.
        // To disable message trace, use:
        // new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(),
        //     new AllocateMessageQueueAveragely());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
            "YOUR GROUP ID", getAclRPCHook(),
            new AllocateMessageQueueAveragely(), true, null);

        // Set the endpoint from the ApsaraMQ for RocketMQ console.
        consumer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");

        // Required for message trace on Alibaba Cloud.
        consumer.setAccessChannel(AccessChannel.CLOUD);

        // Subscribe to all tags (*) on the topic.
        consumer.subscribe("YOUR TOPIC", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("Received messages: %s %n", msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}