Código de exemplo usado para conectar o SDK do cliente TCP RocketMQ 1.x para Java a uma instância do ApsaraMQ for RocketMQ 5.x para enviar e receber mensagens - ApsaraMQ for RocketMQ: Central de Documentação do Alibaba Cloud

As instâncias do ApsaraMQ for RocketMQ 5.x são compatíveis com o SDK do cliente TCP RocketMQ 1.x para Java. O código de exemplo a seguir abrange quatro tipos de mensagem: normal, ordenada, agendada/atrasada e transacional.

Importante

Os SDKs mais recentes do RocketMQ 5.x oferecem compatibilidade total com brokers 5.x e mais recursos. Use-os em todos os novos projetos. Para obter mais informações, consulte Release notes. A Alibaba Cloud mantém apenas os SDKs do cliente TCP e das versões 3.x e 4.x do RocketMQ. Utilize-os exclusivamente para cargas de trabalho existentes.

Índice de códigos

Use a tabela abaixo para acessar diretamente o código necessário.

Tipo de mensagem	Envio	Recebimento
Normal	Síncrono	Push
Normal	Assíncrono	Push em lote
Normal	Unidirecional	Pull
Normal	Multithread
Ordenada	Enviar	Inscrever-se
Agendada	Enviar	Igual ao normal
Atrasada	Enviar	Igual ao normal
Transacional	Enviar	Igual ao normal

Pré-requisitos

Antes de começar, verifique se você tem:

Uma instância do ApsaraMQ for RocketMQ 5.x com endpoint, tópicos e grupos de consumidores criados no console
O SDK do cliente TCP RocketMQ 1.x para Java adicionado ao seu projeto. Para obter mais informações, consulte Preparations

Configuração comum

Todos os exemplos compartilham a mesma configuração de conexão e autenticação mostrada abaixo. Os blocos de código subsequentes omitem esse código base e focam na lógica específica da mensagem.

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;

Properties properties = new Properties();

// Instance credentials
// Obtain the username and password on the Intelligent Authentication tab
// of the Access Control page for your instance in the ApsaraMQ for RocketMQ console.
// Do not use your Alibaba Cloud account AccessKey pair.
properties.put(PropertyKeyConst.AccessKey, "<instance-username>");
properties.put(PropertyKeyConst.SecretKey, "<instance-password>");

// Endpoint
// Enter the domain name and port from the console (e.g., rmq-cn-XXXX.rmq.aliyuncs.com:8080).
// Do not add an http:// or https:// prefix. Do not use a resolved IP address.
properties.put(PropertyKeyConst.NAMESRV_ADDR, "<endpoint>");

// Send timeout in milliseconds
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");

Substitua os seguintes espaços reservados pelos valores reais:

Espaço reservado	Descrição	Exemplo
`<instance-username>`	Nome de usuário da instância obtido no console	LTAI5tXxx
`<instance-password>`	Senha da instância obtida no console	xXxXxXx
`<endpoint>`	Endpoint da instância obtido no console	rmq-cn-XXXX.rmq.aliyuncs.com:8080

Regras de autenticação

Método de acesso	Credenciais obrigatórias?
Endpoint público	Sim. Defina `AccessKey` e `SecretKey`
Virtual Private Cloud (vpc) a partir de uma instância do Elastic Compute Service (ECS)	Não. O broker recupera as credenciais automaticamente das informações da vpc
Instância serverless pela Internet	Sim
Instância serverless em uma vpc (autenticação livre ativada)	Não

Não defina o ID da instância ao usar um SDK de cliente TCP para se conectar a uma instância 5.x. Essa definição causa falhas na conexão.

Instâncias serverless pela Internet

Se você acessar uma instância serverless pela Internet, use a versão 1.9.0.Final ou posterior do SDK e adicione a seguinte propriedade:

properties.setProperty(PropertyKeyConst.Namespace, "<instance-id>");

Substitua <instance-id> pelo ID da sua instância do ApsaraMQ for RocketMQ.

Enviar e receber mensagens normais

Enviar mensagens normais de forma síncrona

O envio síncrono bloqueia a execução até que o broker confirme o recebimento da mensagem. Use este modo quando a confirmação de entrega for essencial.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Date;
import java.util.Properties;

public class ProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        // Call start() once before sending any messages.
        producer.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message(
                    "TopicTestMQ",           // Topic created in the console.
                                             // A normal message topic cannot send or receive other message types.
                    "TagA",                  // Tag for consumer-side filtering.
                    "Hello MQ".getBytes());  // Message body (binary). Producer and consumer must
                                             // agree on serialization and deserialization methods.

            // Business key. Should be globally unique when possible.
            // Use it to look up messages in the console if delivery issues occur.
            msg.setKey("ORDERID_" + i);

            try {
                SendResult sendResult = producer.send(msg);
                // No exception means the message was sent successfully.
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                // Implement retry or persistence logic here.
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting the application.
        // Skip this if the application sends messages continuously.
        producer.shutdown();
    }
}

Enviar mensagens normais de forma assíncrona

O envio assíncrono retorna imediatamente e entrega o resultado por meio de um callback. Este modo é ideal para obter maior throughput quando a thread chamadora não deve ser bloqueada.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class ProducerTest {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "TopicTestMQ",
                "TagA",
                "Hello MQ".getBytes());

        msg.setKey("ORDERID_100");

        // sendAsync returns immediately. The result is delivered through the callback.
        producer.sendAsync(msg, new SendCallback() {
            @Override
            public void onSuccess(final SendResult sendResult) {
                System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
            }

            @Override
            public void onException(OnExceptionContext context) {
                // Implement retry or persistence logic here.
                System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());
            }
        });

        // Wait for the async callback before shutting down.
        TimeUnit.SECONDS.sleep(3);

        producer.shutdown();
    }
}

Enviar mensagens normais no modo unidirecional

O envio unidirecional dispara a mensagem sem aguardar uma resposta do broker. Esse modo oferece o maior throughput possível, mas não garante a entrega. Use-o apenas em cenários que toleram perda ocasional de mensagens, como coleta de logs.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message(
                    "TopicTestMQ",
                    "TagA",
                    "Hello MQ".getBytes());

            msg.setKey("ORDERID_" + i);

            // sendOneway does not wait for a broker response.
            // If delivery confirmation matters, use synchronous or asynchronous sending instead.
            producer.sendOneway(msg);
        }

        producer.shutdown();
    }
}

Enviar mensagens normais a partir de múltiplas threads

O produtor é seguro para threads. Compartilhe uma única instância do produtor entre as threads em vez de criar uma instância por thread.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.util.Date;
import java.util.Properties;

public class SharedProducer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        // Both threads share the same producer instance.
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                Message msg = new Message(
                        "TopicTestMQ",
                        "TagA",
                        "Hello MQ".getBytes());
                try {
                    SendResult sendResult = producer.send(msg);
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }
        });
        thread.start();

        Thread anotherThread = new Thread(new Runnable() {
            @Override
            public void run() {
                Message msg = new Message("TopicTestMQ", "TagA", "Hello MQ".getBytes());
                try {
                    SendResult sendResult = producer.send(msg);
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }
        });
        anotherThread.start();

        // (Optional) Shut down the producer when it is no longer needed.
        // producer.shutdown();
    }
}

Inscrever-se em mensagens normais no modo push

No modo push, o broker entrega as mensagens aos consumidores assim que elas chegam. Este é o padrão de consumo mais comum.

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ConsumerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Consumption mode (optional):
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);    // Default
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

        Consumer consumer = ONSFactory.createConsumer(properties);

        // Subscribe to multiple tags with "||".
        consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });

        // Subscribe to another topic. Use "*" to receive all tags.
        // To unsubscribe, remove the subscription code and restart the consumer.
        consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });

        consumer.start();
        System.out.println("Consumer Started");
    }
}

Inscrever-se em mensagens normais no modo push em lote

O modo push em lote entrega várias mensagens por invocação de callback, reduzindo a sobrecarga por mensagem individual.

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        consumerProperties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Maximum number of messages per batch. Valid values: 1 to 1024. Default: 32.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Maximum wait time between batches in seconds. Valid values: 0 to 450. Default: 0.
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe("TopicTestMQ", "TagA", new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process multiple messages at a time.
                return Action.CommitMessage;
            }
        });

        batchConsumer.start();
        System.out.println("Consumer start success.");

        // Keep the process alive.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Inscrever-se em mensagens normais no modo pull

No modo pull, o consumidor controla quando buscar as mensagens. Use este modo quando precisar de controle preciso sobre o ritmo de consumo.

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.PullConsumer;
import com.aliyun.openservices.ons.api.TopicPartition;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class PullConsumerClient {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-xxxxx");
        // ... common configuration (see above) ...

        PullConsumer consumer = ONSFactory.createPullConsumer(properties);
        consumer.start();

        // Get all partitions for the topic.
        Set<TopicPartition> topicPartitions = consumer.topicPartitions("topic-xxx");
        // Assign partitions to pull from.
        consumer.assign(topicPartitions);

        while (true) {
            // Poll with a 3000 ms timeout.
            List<Message> messages = consumer.poll(3000);
            System.out.printf("Received message: %s %n", messages);
        }
    }
}

Enviar e receber mensagens ordenadas

Mensagens ordenadas garantem que as mensagens com a mesma chave de fragmentação sejam consumidas na ordem em que foram enviadas. Use-as em cenários que exigem sequenciamento estrito, como atualizações de status de pedidos.

Enviar mensagens ordenadas

Use OrderProducer e forneça uma chave de fragmentação para rotear mensagens relacionadas para a mesma partição.

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.order.OrderProducer;

import java.util.Date;
import java.util.Properties;

public class ProducerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        OrderProducer producer = ONSFactory.createOrderProducer(properties);
        producer.start();

        for (int i = 0; i < 1000; i++) {
            String orderId = "biz_" + i % 10;
            Message msg = new Message(
                    "Order_global_topic",              // Topic for ordered messages.
                    "TagA",
                    "send order global msg".getBytes()
            );
            msg.setKey(orderId);

            // Sharding key determines partition routing.
            // Messages with the same sharding key are delivered to the same partition
            // and consumed in order.
            String shardingKey = String.valueOf(orderId);
            try {
                SendResult sendResult = producer.send(msg, shardingKey);
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

Inscrever-se em mensagens ordenadas

Use OrderConsumer com um MessageOrderListener. Retorne OrderAction.Suspend se o consumo falhar, para que o broker tente reenviar a mensagem sem quebrar a ordem.

package com.aliyun.openservices.ons.example.order;

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.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;

import java.util.Properties;

public class ConsumerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Retry interval (ms) if consumption fails. Valid values: 10 to 30000.
        properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
        // Maximum retry count for failed messages.
        properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");

        OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);

        consumer.subscribe(
                "Order_global_topic",
                // Tag filter:
                // "*"           -> subscribe to all tags
                // "TagA||TagB"  -> subscribe to TagA or TagB
                "*",
                new MessageOrderListener() {
                    @Override
                    public OrderAction consume(Message message, ConsumeOrderContext context) {
                        System.out.println(message);
                        // Return OrderAction.Suspend if consumption fails or an exception occurs.
                        return OrderAction.Success;
                    }
                });

        consumer.start();
    }
}

Enviar e receber mensagens agendadas

Mensagens agendadas especificam um horário absoluto de entrega em vez de serem enviadas imediatamente. Elas são adequadas para tarefas que devem ocorrer em um momento preciso, como o envio de um lembrete 24 horas após o registro do usuário.

Enviar mensagens agendadas

Defina startDeliverTime como um timestamp Unix em milissegundos. O broker entregará a mensagem nesse horário. Se o timestamp estiver no passado, o broker entregará a mensagem imediatamente.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "Topic",
                "tag",
                "Hello MQ".getBytes());
        msg.setKey("ORDERID_100");

        try {
            // Deliver at a specific time. Replace with your target timestamp.
            long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-03-07 16:21:00").getTime();
            msg.setStartDeliverTime(timeStamp);

            SendResult sendResult = producer.send(msg);
            System.out.println("Message Id:" + sendResult.getMessageId());
        } catch (Exception e) {
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }

        producer.shutdown();
    }
}

Inscrever-se em mensagens agendadas

A inscrição em mensagens agendadas segue o mesmo processo das mensagens normais. Consulte Inscrever-se em mensagens normais no modo push.

Enviar e receber mensagens atrasadas

Mensagens atrasadas especificam um atraso relativo a partir do momento atual, em vez de um horário absoluto de entrega. São úteis para cenários como a nova tentativa de uma operação falha após um período de espera.

Enviar mensagens atrasadas

Defina startDeliverTime como System.currentTimeMillis() somado ao atraso desejado em milissegundos. O atraso máximo é de 40 dias (3.456.000.000 ms).

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "Topic",
                "tag",
                "Hello MQ".getBytes());
        msg.setKey("ORDERID_100");

        try {
            // Deliver 3 seconds from now.
            long delayTime = System.currentTimeMillis() + 3000;
            msg.setStartDeliverTime(delayTime);

            SendResult sendResult = producer.send(msg);
            if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
            }
        } catch (Exception e) {
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }

        producer.shutdown();
    }
}

Inscrever-se em mensagens atrasadas

A inscrição em mensagens atrasadas segue o mesmo processo das mensagens normais. Consulte Inscrever-se em mensagens normais no modo push.

Enviar e receber mensagens transacionais

Mensagens transacionais vinculam uma transação local e a entrega da mensagem para que ambas tenham sucesso ou falhem como uma unidade. O fluxo de trabalho é:

O broker envia uma meia-mensagem.
Sua aplicação executa a transação local.
Com base no resultado da transação, o broker confirma ou reverte a mensagem.

Se o broker não receber uma confirmação ou reversão a tempo, ele chamará um callback de verificação para consultar o status da transação.

Mensagens transacionais exigem um grupo de consumidores dedicado. Não compartilhe o grupo com outros tipos de mensagem.

Enviar mensagens transacionais

Implemente duas interfaces:

LocalTransactionExecuter: Executa sua transação local após o broker aceitar a meia-mensagem.
LocalTransactionChecker: Responde às verificações de status do broker quando o resultado da transação é desconhecido.

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();
        // Transactional messages require a dedicated group ID.
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Register the transaction status checker before creating the producer.
        LocalTransactionCheckerImpl localTransactionChecker = new LocalTransactionCheckerImpl();
        TransactionProducer transactionProducer = ONSFactory.createTransactionProducer(properties, localTransactionChecker);
        transactionProducer.start();

        Message msg = new Message("XXX", "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 the local transaction here.
                        System.out.println("Execute the local transaction and commit the transaction status.");
                        return TransactionStatus.CommitTransaction;
                    }
                }, null);
                assert sendResult != null;
            } catch (ONSClientException e) {
                System.out.println(new Date() + " Send mq message failed! Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

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

// Transaction status checker.
// The broker calls check() when it does not receive a commit or rollback.
class LocalTransactionCheckerImpl implements LocalTransactionChecker {

    @Override
    public TransactionStatus check(Message msg) {
        System.out.println("The request to check the transaction status of the message is received. MsgId: " + msg.getMsgID());
        return TransactionStatus.CommitTransaction;
    }
}

Inscrever-se em mensagens transacionais

A inscrição em mensagens transacionais segue o mesmo processo das mensagens normais. Consulte Inscrever-se em mensagens normais no modo push.

Próximos passos

Configure logs no lado do cliente para ajudar na solução de problemas de entrega de mensagens. Consulte Configurações de log.
Migre para os SDKs mais recentes do RocketMQ 5.x para obter mais recursos e melhor compatibilidade. Consulte Release notes.