ApsaraMQ for RabbitMQ:Envio e recebimento de mensagens usando um SDK

Pré-requisitos

• Crie os recursos necessários. Criar recursos

• Instale o IntelliJ IDEA

  Use o IntelliJ IDEA ou o Eclipse. Este tópico utiliza o IntelliJ IDEA Ultimate.

• Instale o JDK

• Instale o Maven

Obtenha o endpoint da instância

Especifique o endpoint usado por produtores e consumidores para acessar a instância do ApsaraMQ for RabbitMQ.

1. Faça login no console do ApsaraMQ for RabbitMQ. No painel de navegação à esquerda, clique em Instances.

2. Na barra de navegação superior da página Instances, selecione a região da instância que deseja gerenciar. Em seguida, na lista de instâncias, clique no nome da instância.

3. Na aba Endpoint Information da página Instance Details, passe o mouse sobre o endpoint desejado e copie-o ao clicar no ícone ao lado dele.

   Tipo	Descrição	Exemplo
   Endpoint público	Acesse a instância pela Internet. Instâncias de pagamento conforme o uso aceitam endpoints públicos por padrão. Para instâncias de assinatura, ative o acesso à Internet durante a criação.	XXX.net.mq.amqp.aliyuncs.com
   Endpoint de VPC	Acesse a instância a partir de uma VPC. Instâncias de pagamento conforme o uso e de assinatura aceitam endpoints de VPC por padrão.	XXX.vpc.mq.amqp.aliyuncs.com

   Nota

   Se a sua instância usar autenticação e autorização de código aberto, algumas edições exigem que você também especifique o ID da instância em AMQP Client Properties ao conectar-se via endpoint público. A ausência dessa configuração resulta em falha de autenticação (AccessAccountDataError). Para mais informações, consulte @xref node="{new_doc_nodeId}" Acessar instâncias de autenticação de código aberto por meio de um endpoint público @/xref .

Adicione a dependência Java

1. Crie um projeto Java no IntelliJ IDEA.

2. Adicione a seguinte dependência ao arquivo pom.xml:

   <dependency>
       <groupId>com.rabbitmq</groupId>
       <artifactId>amqp-client</artifactId>
       <version>5.5.0</version> <!-- All versions of open-source RabbitMQ are supported. -->
   </dependency>

Crie um nome de usuário e uma senha

O SDK do RabbitMQ de código aberto exige nome de usuário e senha para autenticação. Gere as credenciais com um dos métodos a seguir e configure os parâmetros userName e passWord no seu código.

• Autenticação e autorização de código aberto

  1. Faça login no console do ApsaraMQ for RabbitMQ. No painel de navegação à esquerda, clique em Instances.

  2. Na barra de navegação superior da página Instances, selecione a região da instância que deseja gerenciar. Em seguida, na lista de instâncias, clique no nome da instância.

  3. No painel de navegação à esquerda, clique em Users and Permissions.

  4. Na página Users and Permissions, clique em Create Username/Password.

  5. No painel Create Username/Password, preencha os campos Username, Password e Confirm Password e, em seguida, clique em OK.

  Nota

  Após criar as credenciais, conceda permissões ao usuário. Gerenciar usuários e permissões. Se conectar-se à instância via endpoint público, consulte @xref node="{new_doc_nodeId}" Acessar instâncias de autenticação de código aberto por meio de um endpoint público @/xref para ver a configuração adicional necessária.

• Resource Access Management (RAM)

  1. Faça login no console do ApsaraMQ for RabbitMQ. No painel de navegação à esquerda, clique em Instances.

  2. Na barra de navegação superior da página Instances, selecione a região da instância que deseja gerenciar. Em seguida, na lista de instâncias, clique no nome da instância.

  3. No painel de navegação à esquerda, clique em Users and Permissions.

  4. Na página Users and Permissions, clique em Create Username/Password.

  5. No painel Create Username/Password, insira o AccessKey ID e o AccessKey Secret e, em seguida, clique em OK.

     Nota

     Obtenha o AccessKey ID e o AccessKey Secret no console do RAM da Alibaba Cloud. Para mais informações, consulte Criar um par de AccessKey.

     Importante

     O ApsaraMQ for RabbitMQ não aceita o uso direto de um AccessKey da Alibaba Cloud (AccessKey ID e AccessKey Secret) como credencial de conexão do cliente. Primeiro, converta o AccessKey em credenciais estáticas no console e, depois, use o nome de usuário e a senha gerados para autenticar a conexão do cliente. Se usar o AccessKey diretamente como credencial, a conexão falhará com o erro AuthenticationFailureException: AkDisabled.

     Após criar as credenciais, a página Users and Permissions exibe o novo nome de usuário e a nova senha estáticos, com a senha oculta por padrão. A nova entrada aparece na lista de credenciais com colunas como Instance ID, AccessKey, Username, Password e Creation Time. Clique em Delete na coluna Actions para remover a credencial.

  6. Na coluna Password da credencial estática criada, clique em Display para visualizar a senha.

Etapa 4: Crie a fábrica de conexões

A classe ConnectionFactory gerencia a configuração da conexão, o SSL/TLS e a recuperação automática. Crie um arquivo chamado ConnectionFactory.java:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;

public class ConnectionFactory {
    private final String hostName;
    private final int port;
    private final String userName;
    private final String password;
    private final String virtualHost;
    private final boolean enableSSL;

    public ConnectionFactory(String hostName, int port, String userName,
                            String password, String virtualHost, boolean enableSSL) {
        this.hostName = hostName;
        this.port = port;
        this.userName = userName;
        this.password = password;
        this.virtualHost = virtualHost;
        this.enableSSL = enableSSL;
    }

    public Channel createChannel() throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        Connection con = createCon();
        return con.createChannel();
    }

    private Connection createCon() throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();

        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(password);

        // Reconnect automatically if the connection drops.
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        factory.setPort(port);

        if (enableSSL) {
            setSSL(factory);
        }

        // Adjust these timeouts based on your network conditions.
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);
        factory.setShutdownTimeout(0);

        return factory.newConnection();
    }

    private void setSSL(com.rabbitmq.client.ConnectionFactory factory) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        trustManagerFactory.init((KeyStore) null);
        sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
        factory.useSslProtocol(sslContext);
    }

    public void closeCon(Channel channel) {
        if (channel != null && channel.getConnection() != null) {
            try {
                channel.getConnection().close();
            } catch (Throwable t) {
            }
        }
    }
}

Configurações principais:

Produza mensagens

No seu projeto Java, crie um programa produtor chamado Producer.java, configure os parâmetros conforme descrito em Parâmetros do SDK e execute o programa. Para precauções sobre o envio de mensagens, consulte O que preciso observar ao produzir mensagens?

Código de exemplo:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class Producer {
    // The endpoint of the ApsaraMQ for RabbitMQ instance. 
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    // The static username of the ApsaraMQ for RabbitMQ instance. 
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    // The static password of the ApsaraMQ for RabbitMQ instance. 
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    // The name of the vhost of the ApsaraMQ for RabbitMQ instance. 
    public static final String virtualHost = "vhost_test";

    // If you want to use port 5671, you must set the enableSSL parameter to true. 
    public static final int port = 5672;
    public static final boolean enableSSL = false;

    private Channel channel;
    private final ConcurrentNavigableMap<Long/*deliveryTag*/, String/*msgId*/> outstandingConfirms;
    private final ConnectionFactory factory;
    private final String exchangeName;
    private final String queueName;
    private final String routingKey;

    public Producer(ConnectionFactory factory, String exchangeName, String queueName, String routingKey) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        this.factory = factory;
        this.outstandingConfirms = new ConcurrentSkipListMap<>();
        this.channel = factory.createChannel();
        this.exchangeName = exchangeName;
        this.queueName = queueName;
        this.routingKey = routingKey;
    }

    public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        // Create a connection factory. 
        ConnectionFactory factory = new ConnectionFactory(hostName, port, userName, password, virtualHost, enableSSL);

        // Initialize the producer. 
        Producer producer = new Producer(factory, "ExchangeTest", "QueueTest", "RoutingKeyTest");

        // Declare the producer. 
        producer.declare();

        producer.initChannel();

        // Send messages. 
        producer.doSend("hello,amqp");
    }

    private void initChannel() throws IOException {
        channel.confirmSelect();

        ConfirmCallback cleanOutstandingConfirms = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);

                for (Long tag : confirmed.keySet()) {
                    String msgId = confirmed.get(tag);
                    System.out.format("Message with msgId %s has been ack-ed. deliveryTag: %d, multiple: %b%n", msgId, tag, true);
                }

                confirmed.clear();
            } else {
                String msgId = outstandingConfirms.remove(deliveryTag);
                System.out.format("Message with msgId %s has been ack-ed. deliveryTag: %d, multiple: %b%n", msgId, deliveryTag, false);
            }
        };
        channel.addConfirmListener(cleanOutstandingConfirms, (deliveryTag, multiple) -> {
            String msgId = outstandingConfirms.get(deliveryTag);
            System.err.format("Message with msgId %s has been nack-ed. deliveryTag: %d, multiple: %b%n", msgId, deliveryTag, multiple);
            // send msg failed, re-publish
        });

        channel.addReturnListener(returnMessage -> System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
    }

    private void declare() throws IOException {
        channel.exchangeDeclare(exchangeName, "direct", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
    }
    

    private void doSend(String content) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        try {
            String msgId = UUID.randomUUID().toString();
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

            channel.basicPublish(exchangeName, routingKey, true, props, content.getBytes(StandardCharsets.UTF_8));

            outstandingConfirms.put(channel.getNextPublishSeqNo(), msgId);
        } catch (AlreadyClosedException e) {
            //need reconnect if channel is closed.
            String message = e.getMessage();

            System.out.println(message);

            if (channelClosedByServer(message)) {
                factory.closeCon(channel);
                channel = factory.createChannel();
                this.initChannel();
                doSend(content);
            } else {
                throw e;
            }
        }
    }

    private boolean channelClosedByServer(String errorMsg) {
        if (errorMsg != null
            && errorMsg.contains("channel.close")
            && errorMsg.contains("reply-code=541")
            && errorMsg.contains("reply-text=InternalError")) {
            return true;
        } else {
            return false;
        }
    }
}

Consuma mensagens

No seu projeto Java, crie um programa consumidor chamado Consumer.java, configure os parâmetros conforme descrito em Parâmetros do SDK e execute o programa.

Código de exemplo:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;

public class Consumer {
    // The endpoint of the ApsaraMQ for RabbitMQ instance. 
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    // The static username of the ApsaraMQ for RabbitMQ instance. 
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    // The static password of the ApsaraMQ for RabbitMQ instance. 
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    // The name of the vhost of the ApsaraMQ for RabbitMQ instance. 
    public static final String virtualHost = "vhost_test";
    
    // If you want to use port 5671, you must set the enableSSL parameter to true. 
    public static final int port = 5672;
    public static final boolean enableSSL = false;

    private final Channel channel;
    private final String queue;

    public Consumer(Channel channel, String queue) {
        this.channel = channel;
        this.queue = queue;
    }

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        ConnectionFactory factory = new ConnectionFactory(hostName, port, userName, password, virtualHost, enableSSL);
        Channel channel = factory.createChannel();
        channel.basicQos(50);
         
        // The name of the queue on the ApsaraMQ for RabbitMQ instance. The queue name that you specify for this parameter must be consistent with the queue name that you specified when you created the producer. 
        Consumer consumer = new Consumer(channel, "queue-1");

        consumer.consume();
    }

    public void consume() throws IOException, InterruptedException {
        channel.basicConsume(queue, false, new DefaultConsumer(channel) {
            @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {

                // Process the message. 
                System.out.println("receive: msgId=" + properties.getMessageId());

                // The consumer must commit acknowledgment (ACK) within the validity period. Otherwise, the message is pushed again. The message can be pushed up to 16 times. 
                // If the message still fails to be pushed after 16 times, it is discarded or sent to the dead-letter exchange. 
                // The validity period is 1 minute for Professional Edition instances, 5 minutes for Enterprise Edition instances, 5 minutes for Serverless shared architecture instances, 30 minutes for Serverless dedicated architecture instances, and 30 minutes for Enterprise Platinum Edition instances. 
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });

        CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                channel.getConnection().close();
            } catch (IOException e) {
                System.out.println("close connection error." + e);
            }
            latch.countDown();
        }));
        latch.await();
    }
}

Parâmetros do SDK

Referências

• O ApsaraMQ for RabbitMQ é totalmente compatível com o RabbitMQ de código aberto e aceita SDKs para diversas linguagens. Para informações sobre SDKs de outras linguagens de programação, consulte SDK de linguagem ou framework Java Biblioteca de cliente Java do RabbitMQ Spring Framework Projeto Spring AMQP para Java .NET SDK do .NET Python SDK do Python PHP SDK do PHP Rust SDK do Rust C e C++ SDK de C e C++ Go SDK do Go JavaScript e Node SDK de JavaScript e Node Objective-C e Swift SDK de Objective-C e Swift. Para outros parâmetros, consulte a Documentação do cliente.

• Para solucionar erros de cliente, consulte Códigos de erro AMQP.

• Pesquise mensagens e rastros no console do ApsaraMQ for RabbitMQ para verificar a entrega de mensagens. Para mais informações, consulte Mensagens e Rastros de mensagens.