Sample code used to connect RocketMQ 3.x or 4.x SDK for Java to an ApsaraMQ for RocketMQ 5.x instance to send and receive messages - ApsaraMQ for RocketMQ: Central de Documentação do Alibaba Cloud

As instâncias do ApsaraMQ for RocketMQ 5.x são compatíveis com clientes desenvolvidos com os SDKs do RocketMQ 3.x e 4.x. Esta página apresenta exemplos em Java para envio e recebimento de mensagens normais, ordenadas, agendadas, atrasadas e transacionais usando esses SDKs legados.

Importante

Os SDKs mais recentes do RocketMQ 5.x oferecem compatibilidade total com os brokers do ApsaraMQ for RocketMQ 5.x e incluem recursos adicionais. Utilize os SDKs 5.x para novos projetos. Para mais detalhes, consulte Release notes. Os SDKs cliente 3.x, 4.x e TCP recebem manutenção apenas para cargas de trabalho existentes.

Pré-requisitos

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

Uma instância do ApsaraMQ for RocketMQ 5.x
Um tópico e um grupo de consumidores criados no console do ApsaraMQ for RocketMQ
O SDK do RocketMQ 3.x ou 4.x para Java adicionado ao seu projeto
O ponto de acesso da instância, no formato rmq-cn-XXXX.rmq.aliyuncs.com:8080 (disponível no console)

Para obter instruções completas de configuração, consulte Preparations.

Configuração compartilhada

Todos os exemplos desta página compartilham a mesma configuração de autenticação e conexão. Os blocos a seguir definem essa configuração comum, permitindo que cada exemplo de tipo de mensagem foque na lógica de envio ou recebimento.

Autenticação

O ApsaraMQ for RocketMQ oferece dois métodos de acesso. Escolha aquele adequado ao seu ambiente de rede:

Endpoint público -- Configure o RPCHook com o nome de usuário e a senha da instância. Obtenha essas credenciais na aba Intelligent Authentication, dentro da página Access Control no console do ApsaraMQ for RocketMQ. Não utilize o par de AccessKey da sua conta Alibaba Cloud.
Endpoint de VPC -- Credenciais não são necessárias. Quando o cliente é executado em uma instância do Elastic Compute Service (ECS) dentro de uma Virtual Private Cloud (VPC), o broker obtém automaticamente as credenciais da configuração da VPC.

Para instâncias serverless, as credenciais são obrigatórias no acesso via endpoint público. Caso você ative o recurso de acesso sem autenticação em VPCs, o acesso pela VPC dispensa credenciais.

Auxiliar RPCHook

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.remoting.RPCHook;

/**
 * Returns an RPCHook configured with instance credentials.
 * Required for public endpoint access and serverless instances over the Internet.
 * Not required for VPC access on standard instances.
 */
private static RPCHook getAclRPCHook() {
    return new AclClientRPCHook(new SessionCredentials(
        "<instance-username>",   // From the Intelligent Authentication tab
        "<instance-password>"    // From the Intelligent Authentication tab
    ));
}

Inicialização do produtor

import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

// --- Public endpoint: pass RPCHook ---
DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());

// --- VPC endpoint: omit RPCHook ---
// DefaultMQProducer producer = new DefaultMQProducer();

// Group ID created in the console
producer.setProducerGroup("<your-group-id>");

// Required for message trace. Omit to disable tracing.
producer.setAccessChannel(AccessChannel.CLOUD);

// Access point from the console. Use the domain and port only -- no http:// prefix, no resolved IP.
producer.setNamesrvAddr("<your-access-point>");

producer.start();

Inicialização do consumidor

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

// --- Public endpoint: pass RPCHook ---
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());

// --- VPC endpoint: omit RPCHook ---
// DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();

consumer.setConsumerGroup("<your-group-id>");
consumer.setAccessChannel(AccessChannel.CLOUD);
consumer.setNamesrvAddr("<your-access-point>");
consumer.subscribe("<your-topic>", "*");

Referência de placeholders

Substitua os placeholders abaixo pelos seus valores reais:

Placeholder	Descrição	Exemplo
`<instance-username>`	Nome de usuário da instância obtido na aba Intelligent Authentication	`MjY1NTI5MD...`
`<instance-password>`	Senha da instância obtida na aba Intelligent Authentication	`OTk2QjFBMz...`
`<your-group-id>`	ID do grupo de consumidores criado no console	`GID_test_group`
`<your-access-point>`	Ponto de acesso da instância (domínio:porta) disponível no console	`rmq-cn-hangzhou.rmq.aliyuncs.com:8080`
`<your-topic>`	Tópico criado no console	`test_topic`

Mensagens normais

As mensagens normais suportam três modos de envio: síncrono, assíncrono e unidirecional. Para detalhes conceituais, consulte Normal messages.

Enviar mensagens normais de forma síncrona

O envio síncrono bloqueia a execução até que o broker confirme cada mensagem. Recomenda-se este modo quando a confirmação de entrega for obrigatória.

import java.util.Date;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// After producer initialization (see Shared configuration)

for (int i = 0; i < 128; i++) {
    try {
        Message msg = new Message(
            "<your-topic>",
            "<your-message-tag>",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
    } catch (Exception e) {
        // Handle send failure: retry or persist the message
        System.out.println(new Date() + " Send mq message failed.");
        e.printStackTrace();
    }
}

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

Enviar mensagens normais de forma assíncrona

O envio assíncrono retorna imediatamente e entrega os resultados por meio de um callback. Este modo é ideal para produtores sensíveis à latência.

import java.util.Date;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// After producer initialization (see Shared configuration)

for (int i = 0; i < 128; i++) {
    try {
        Message msg = new Message(
            "<your-topic>",
            "<your-message-tag>",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult result) {
                System.out.println("send message success. msgId= " + result.getMsgId());
            }

            @Override
            public void onException(Throwable throwable) {
                // Handle send failure: retry or persist the message
                System.out.println("send message failed.");
                throwable.printStackTrace();
            }
        });
    } catch (Exception e) {
        System.out.println(new Date() + " Send mq message failed.");
        e.printStackTrace();
    }
}

// Wait for async callbacks to complete before shutting down
TimeUnit.SECONDS.sleep(3);
producer.shutdown();

Enviar mensagens normais no modo unidirecional

O envio unidirecional dispara a mensagem sem aguardar resposta do broker. Utilize este modo para cargas de trabalho de alto throughput nas quais a perda eventual de mensagens seja aceitável, como na coleta de logs.

import java.util.Date;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// After producer initialization (see Shared configuration)

for (int i = 0; i < 128; i++) {
    try {
        Message msg = new Message(
            "<your-topic>",
            "<your-message-tag>",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

        producer.sendOneway(msg);
    } catch (Exception e) {
        System.out.println(new Date() + " Send mq message failed.");
        e.printStackTrace();
    }
}

producer.shutdown();

Receber mensagens normais

Inscreva-se em mensagens normais com um listener concorrente. Cada mensagem é processada independentemente, sem garantia de ordenação.

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.common.message.MessageExt;
import java.util.List;

// After consumer initialization (see Shared configuration)

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

Mensagens ordenadas

As mensagens ordenadas garantem entrega FIFO (first-in-first-out) dentro da mesma fila de mensagens. Mensagens com a mesma chave de sharding são roteadas para a mesma fila. Para detalhes conceituais, consulte Ordered messages.

Enviar mensagens ordenadas

Diferentemente das mensagens normais, as mensagens ordenadas exigem um MessageQueueSelector para rotear mensagens com a mesma chave de sharding para a mesma fila.

import java.util.List;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// After producer initialization (see Shared configuration)

for (int i = 0; i < 128; i++) {
    try {
        int orderId = i % 10;
        Message msg = new Message(
            "<your-order-topic>",
            "<your-message-tag>",
            "OrderID188",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // Set the sharding key to ensure ordered delivery.
        // For ApsaraMQ for RocketMQ 5.x instances, an alternative is:
        // msg.putUserProperty(MessageConst.PROPERTY_SHARDING_KEY, orderId + "");
        msg.putUserProperty("__SHARDINGKEY", orderId + "");

        SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // Route messages with the same orderId to the same queue
                Integer id = (Integer) arg;
                int index = id % mqs.size();
                return mqs.get(index);
            }
        }, orderId);

        System.out.printf("%s%n", sendResult);
    } catch (Exception e) {
        e.printStackTrace();
    }
}
producer.shutdown();

Receber mensagens ordenadas

Utilize MessageListenerOrderly em vez de MessageListenerConcurrently. O broker entrega as mensagens em ordem dentro de cada fila.

import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

// After consumer initialization (see Shared configuration)
// Subscribe to the ordered topic instead of a normal topic:
// consumer.subscribe("<your-order-topic>", "*");

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                ConsumeOrderlyContext context) {
        context.setAutoCommit(true);
        System.out.printf("%s Receive New Messages: %s %n",
            Thread.currentThread().getName(), msgs);

        // Return SUSPEND_CURRENT_QUEUE_A_MOMENT on failure to retry
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

consumer.start();
System.out.printf("Consumer Started.%n");

Mensagens agendadas e atrasadas

Mensagens agendadas são entregues em um horário específico. Mensagens atrasadas são entregues após um intervalo definido. Ambos os tipos utilizam a propriedade __STARTDELIVERTIME. Para detalhes conceituais, consulte Scheduled and delayed messages.

Enviar mensagens agendadas ou atrasadas

import java.util.Date;
import java.text.SimpleDateFormat;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// After producer initialization (see Shared configuration)

for (int i = 0; i < 128; i++) {
    try {
        Message msg = new Message(
            "<your-topic>",
            "<your-message-tag>",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // Option 1: Delayed delivery -- deliver 3 seconds from now
        long delayTime = System.currentTimeMillis() + 3000;
        msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(delayTime));

        // Option 2: Scheduled delivery -- deliver at a specific time
        // Format: yyyy-MM-dd HH:mm:ss
        // If the time is in the past, the message is delivered immediately.
        // long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        //     .parse("2021-08-10 18:45:00").getTime();
        // msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(timeStamp));

        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
    } catch (Exception e) {
        System.out.println(new Date() + " Send mq message failed.");
        e.printStackTrace();
    }
}

producer.shutdown();

Receber mensagens agendadas ou atrasadas

O código do consumidor é idêntico ao descrito em Receber mensagens normais. Nenhuma configuração especial é necessária no lado do consumidor.

Mensagens transacionais

As mensagens transacionais fornecem suporte a transações distribuídas por meio de um commit em duas fases: o produtor envia uma meia mensagem, executa uma transação local e então faz o commit ou rollback da mensagem com base no resultado da transação. Para detalhes conceituais, consulte Transactional messages.

Importante

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

Enviar mensagens transacionais

Este exemplo envia uma meia mensagem e executa uma transação local. O broker retém a meia mensagem até que o produtor faça o commit ou rollback.

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.common.RemotingHelper;

// --- Public endpoint: pass RPCHook and a dedicated transaction group ID ---
TransactionMQProducer transactionMQProducer = new TransactionMQProducer(
    "<your-transaction-group-id>", getAclRPCHook());

// --- VPC endpoint: omit RPCHook ---
// TransactionMQProducer transactionMQProducer = new TransactionMQProducer(
//     "<your-transaction-group-id>");

transactionMQProducer.setAccessChannel(AccessChannel.CLOUD);
transactionMQProducer.setNamesrvAddr("<your-access-point>");

// Register the transaction status checker (see next section)
transactionMQProducer.setTransactionCheckListener(new LocalTransactionCheckerImpl());
transactionMQProducer.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));

        SendResult sendResult = transactionMQProducer.sendMessageInTransaction(
            message,
            new LocalTransactionExecuter() {
                @Override
                public LocalTransactionState executeLocalTransactionBranch(
                        Message msg, Object arg) {
                    System.out.println("Start executing the local transaction: " + msg);
                    // Run your local transaction logic here.
                    // Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW.
                    return LocalTransactionState.UNKNOW;
                }
            },
            null);

        assert sendResult != null;
    } catch (Exception e) {
        e.printStackTrace();
    }
}

Verificar status da transação

Quando o produtor retorna UNKNOW, o broker chama periodicamente checkLocalTransactionState para resolver a transação. Implemente a interface TransactionCheckListener para consultar o estado da sua transação local e retornar o resultado final.

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("Transaction status check received. MsgId: " + msg.getMsgId());
        // Query your local data store to determine the transaction outcome.
        // Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW.
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

Receber mensagens transacionais

O código do consumidor é idêntico ao descrito em Receber mensagens normais. Nenhuma configuração especial é necessária no lado do consumidor.

Referências

Preparations -- Configure seu ambiente de desenvolvimento
Overview -- Códigos de exemplo para outras linguagens de programação
Para informações sobre tipos de mensagens, consulte os seguintes tópicos: