Sample code for sending and receiving messages with the Apache RocketMQ Java SDK - ApsaraMQ for RocketMQ: Central de Documentação do Alibaba Cloud

Envie e receba mensagens com o ApsaraMQ for RocketMQ usando o Apache RocketMQ Java SDK. Esta página aborda tanto o SDK do protocolo gRPC (rocketmq-client-java) quanto o SDK do protocolo Remoting (rocketmq-client).

Pré-requisitos

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

Uma instância do ApsaraMQ for RocketMQ com tópicos e group IDs criados no console do ApsaraMQ for RocketMQ
O endpoint da instância (por exemplo, rmq-cn-XXXX.rmq.aliyuncs.com:8080)
A dependência necessária do SDK adicionada ao seu projeto. Para detalhes sobre versões, consulte o Guia de versões

Importante

Se você utilizar uma instância Serverless, verifique os requisitos de versão do SDK para acesso via rede pública antes de prosseguir.

SDK do protocolo gRPC

O SDK rocketmq-client-java se comunica pelo protocolo gRPC. As tabelas a seguir listam códigos de exemplo para cada tipo de mensagem, hospedados no repositório de clientes do Apache RocketMQ.

Para integração com Spring Boot, consulte rocketmq-v5-client-spring-boot-samples.

Importante

Ao enviar mensagens transacionais com o SDK do protocolo gRPC, defina um tópico ao iniciar o producer. Sem um tópico, as verificações de transação sofrem atraso. Se a mensagem não for enviada dentro de quatro horas, a mensagem de meia transação poderá ser descartada.

Enviar mensagens

Tipo de mensagem	Código de exemplo
Mensagem normal (síncrona)	ProducerNormalMessageExample.java
Mensagem normal (assíncrona)	AsyncProducerExample.java
Mensagem ordenada	ProducerFifoMessageExample.java
Mensagem agendada ou atrasada	ProducerDelayMessageExample.java
Mensagem transacional	ProducerTransactionMessageExample.java
Mensagem leve	LiteProducerExample.java

Consumir mensagens

Tipo de consumer	Código de exemplo
Push consumer	PushConsumerExample.java
Simple consumer (síncrono)	SimpleConsumerExample.java
Simple consumer (assíncrono)	AsyncSimpleConsumerExample.java
Lightweight push consumer	LitePushConsumerExample.java

Para mais informações sobre push consumers e simple consumers, consulte Tipos de consumer.

SDK do protocolo Remoting

O SDK rocketmq-client se comunica pelo protocolo Remoting. Esta seção fornece código inline para cada tipo de mensagem.

Para integração com Spring Boot, consulte rocketmq-spring-boot-samples.

Configuração comum

Todos os exemplos do protocolo Remoting compartilham a mesma configuração de autenticação e conexão. Revise esta seção primeiro e depois consulte o código específico para cada tipo de mensagem abaixo.

Autenticação

A forma de inicializar o producer ou consumer depende do método de acesso:

Endpoint público -- Passe um RPCHook com o nome de usuário e a senha da instância: Obtenha o nome de usuário e a senha da instância na aba Intelligent Identity Recognition do console Resource Access Management. Não utilize o AccessKey ID e o AccessKey secret da sua conta Alibaba Cloud.
```
  private static RPCHook getAclRPCHook() {
      return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
  }

  // Pass the RPCHook when creating the producer
  DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
```
Endpoint da VPC -- Nenhum RPCHook é necessário. O servidor autentica com base na VPC:
```
  DefaultMQProducer producer = new DefaultMQProducer();
```
Instância Serverless -- Passe um RPCHook para acesso via rede pública. Se o acesso sem senha pela rede interna estiver ativado, nenhum RPCHook será necessário.

Conexão

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

// Endpoint from the ApsaraMQ for RocketMQ console
// Use the domain name and port as shown. Do not add http:// or https://.
// Do not use a resolved IP address.
producer.setNamesrvAddr("<your-access-point>");

Rastreamento de mensagens (opcional)

Para ativar o rastreamento de mensagens em nuvem, defina o canal de acesso e habilite os rastreamentos:

producer.setAccessChannel(AccessChannel.CLOUD);

// Required for SDK v5.3.0 and later, in addition to setAccessChannel
producer.setEnableTrace(true);

Referência de placeholders

Substitua os placeholders a seguir pelos seus valores reais:

Placeholder	Descrição	Exemplo
`<instance-username>`	Nome de usuário da instância obtido no console	--
`<instance-password>`	Senha da instância obtida no console	--
`<your-group-id>`	Group ID criado no console	GID_example
`<your-access-point>`	Endpoint obtido no console	rmq-cn-XXXX.rmq.aliyuncs.com:8080
`<your-topic>`	Tópico criado no console	topic_example
`<your-order-topic>`	Tópico para mensagens ordenadas	order_topic_example
`<your-transaction-topic>`	Tópico para mensagens transacionais	transaction_topic_example
`<your-transaction-group-id>`	Group ID dedicado para mensagens transacionais	GID_transaction_example
`<your-message-tag>`	Tag de mensagem para filtragem	TagA

Mensagens normais

Mensagens normais são adequadas para a maioria dos casos de uso que não exigem garantias de ordenação ou transação.

Enviar uma mensagem normal (síncrona)

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.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        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) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before the application exits.
        // Note: Destroying the producer object saves system memory.
        // To send messages continuously, keep the producer running.
        producer.shutdown();
    }
}

Enviar uma mensagem normal (assíncrona)

O envio assíncrono retorna imediatamente e invoca um callback quando o broker responde.

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.DefaultMQProducer;
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.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class RocketMQAsyncProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        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) {
                        System.out.println("send message failed.");
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }
        // Wait 3 seconds for the asynchronous callbacks to complete
        TimeUnit.SECONDS.sleep(3);

        producer.shutdown();
    }
}

Enviar uma mensagem normal (unidirecional)

Envios unidirecionais não aguardam resposta do broker. Utilize este modo para cenários de alto throughput onde a perda ocasional de mensagens é aceitável, como na coleta de logs.

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.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQOnewayProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("<your-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // sendOneway does not return a result or throw an exception on failure
                producer.sendOneway(msg);
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

Consumir mensagens normais

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.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.List;

public class RocketMQPushConsumer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());
        consumer.setConsumerGroup("<your-group-id>");
        consumer.setAccessChannel(AccessChannel.CLOUD);
        consumer.setEnableTrace(true);
        consumer.setNamesrvAddr("<your-access-point>");

        // Subscribe to a topic. Use "*" to receive all tags, or specify a tag expression.
        consumer.subscribe("<your-topic>", "*");
        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

Mensagens ordenadas garantem entrega FIFO dentro da mesma chave de sharding. Utilize-as para cenários como processamento sequencial de eventos ou sincronização de dados em tempo real.

Enviar uma mensagem ordenada

Mensagens com a mesma chave de sharding são entregues à mesma fila em ordem. Defina a propriedade __SHARDINGKEY para que o broker direcione mensagens relacionadas à mesma fila.

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.DefaultMQProducer;
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.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

public class RocketMQOrderProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        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 route related messages to the same queue.
                // In v5.x, you can replace the following line with:
                // 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) {
                        // Select a queue based on the orderId
                        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();
    }
}

Consumir mensagens ordenadas

Utilize MessageListenerOrderly em vez de MessageListenerConcurrently para preservar a ordem de consumo dentro de cada fila.

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.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.List;

public class RocketMQOrderConsumer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());
        consumer.setConsumerGroup("<your-group-id>");
        consumer.setAccessChannel(AccessChannel.CLOUD);
        consumer.setEnableTrace(true);
        consumer.setNamesrvAddr("<your-access-point>");
        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 atraso configurável. Ambas utilizam a propriedade de usuário __STARTDELIVERTIME.

Enviar uma mensagem agendada ou atrasada

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.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQDelayProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

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

                // Delayed message: deliver after 3 seconds
                long delayTime = System.currentTimeMillis() + 3000;
                msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(delayTime));

                // Scheduled message: deliver at a specific time (format: yyyy-MM-dd HH:mm:ss)
                // If the specified time is earlier than the current time, 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();
    }
}

Consumir mensagens agendadas e atrasadas

Consuma mensagens agendadas e atrasadas da mesma forma que as mensagens normais. Nenhuma configuração adicional é necessária.

Mensagens transacionais

Mensagens transacionais utilizam um protocolo de commit em duas fases. O producer envia uma mensagem de meia transação, executa uma transação local e então faz o commit ou rollback. O broker verifica periodicamente transações não confirmadas chamando checkLocalTransaction.

Importante

O group ID para mensagens transacionais não pode ser compartilhado com outros tipos de mensagem. Crie um group ID dedicado para producers transacionais.

Enviar uma mensagem transacional

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.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQTransactionProducer {

    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        // Use a dedicated group ID for transactional messages
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("<your-transaction-group-id>", getAclRPCHook());
        transactionMQProducer.setAccessChannel(AccessChannel.CLOUD);
        transactionMQProducer.setEnableTrace(true);
        transactionMQProducer.setNamesrvAddr("<your-access-point>");

        transactionMQProducer.setTransactionListener(new TransactionListener() {

            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // Run your local transaction logic here
                System.out.println("Start to execute the local transaction: " + msg);
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // Return the local transaction status when the broker checks back
                System.out.println("Received a transaction check request, MsgId: " + msg.getMsgId());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        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, null);
                assert sendResult != null;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Consumir mensagens transacionais

Consuma mensagens transacionais da mesma forma que as mensagens normais. Nenhuma configuração adicional é necessária.

Acesso via rede pública em instância Serverless

Para acessar uma instância Serverless pela rede pública, seu SDK deve atender a um requisito de versão mínima. Adicione a configuração de namespace mostrada abaixo.

Nota

Substitua InstanceId pelo ID real da sua instância.

SDK do protocolo Remoting (rocketmq-client >= 5.2.0)

Adicione o namespace ao producer ou consumer:

// Producer
producer.setNamespaceV2("InstanceId");

// Consumer
consumer.setNamespaceV2("InstanceId");

SDK do protocolo gRPC (rocketmq-client-java >= 5.0.6)

Defina o namespace na ClientConfiguration:

ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    .setEndpoints(endpoints)
    .setNamespace("InstanceId")
    .setCredentialProvider(sessionCredentialsProvider)
    .build();

Pré-requisitos

SDK do protocolo gRPC

Enviar mensagens

Consumir mensagens

SDK do protocolo Remoting

Configuração comum

Autenticação

Conexão

Rastreamento de mensagens (opcional)

Referência de placeholders

Mensagens normais

Mensagens ordenadas

Mensagens agendadas e atrasadas

Consumir mensagens agendadas e atrasadas

Mensagens transacionais

Consumir mensagens transacionais

Acesso via rede pública em instância Serverless

SDK do protocolo Remoting (rocketmq-client >= 5.2.0)

SDK do protocolo gRPC (rocketmq-client-java >= 5.0.6)

Consulte também