高弹性，低成本，云消息队列 RabbitMQ实践

说明：云消息队列RabbitMQ版与开源RabbitMQ完全兼容，请关注SDK使用注意事项。更多语言SDK，请参见SDK列表及示例代码。

安装Java依赖库

1. 在IDEA中创建一个Java工程。

2. 在pom.xml文件中添加以下依赖引入Java依赖库。

   <dependency>
       <groupId>com.rabbitmq</groupId>
       <artifactId>amqp-client</artifactId>
       <version>5.5.0</version> 
   </dependency>

生产消息

在已创建的Java工程中，创建消息发送程序，按照代码参数说明部分配置相关参数并运行。

示例代码如下：

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

public class ProducerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        //设置实例的接入点。
        String hostName = "xxx.xxx.aliyuncs.com";
        //设置实例的静态用户名密码。
        String userName = "UserName";
        String passWord = "PassWord";
        //设置实例的Vhost。
        String virtualHost = "test-vhost";

        //在生产环境中，建议提前创建好Connection，并在需要时重复使用，避免频繁创建和关闭Connection，
        // 以提高性能、复用连接资源，以及保证系统的稳定性。
        Connection connection = createConnection(hostName, userName, passWord, virtualHost);
        Channel channel = connection.createChannel();

        //设置Exchange、Queue和绑定关系。
        String exchangeName = "test-exchange";
        String queueName = "test-queue";
        String bindingKey = "test-routing-key";
        //设置Exchange类型。
        String exchangeType = "direct";

        //开始发送消息。
        for (int i = 0; i < 10; i++) {
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(UUID.randomUUID().toString()).build();
            channel.basicPublish(exchangeName, bindingKey, true, props,
                    ("消息发送示例Body-" + i).getBytes(StandardCharsets.UTF_8));
            System.out.println("[SendResult] Message sent successfully, messageId: " + props.getMessageId()
                    + ", exchange: " + exchangeName + ", routingKey: " + bindingKey);
        }
        channel.close();
        connection.close();
    }

    public static Connection createConnection(String hostName, String userName, String passWord, String virtualHost)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(passWord);
        //设置为true,开启Connection自动恢复功能；设置为false,关闭Connection自动恢复功能。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        //默认端口，非加密端口5672，加密端口5671。
        factory.setPort(5672);
        //基于网络环境合理设置超时时间。
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);
        factory.setShutdownTimeout(0);
        Connection connection = factory.newConnection();
        return connection;
    }
}

订阅消息

在已创建的Java工程中，创建消息订阅程序，按照代码参数说明部分配置相关参数并运行

示例代码如下：

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

public class ConsumerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        //设置实例的接入点。
        String hostName = "xxx.xxx.aliyuncs.com";
        //设置实例的静态用户名密码。
        String userName = "UserName";
        String passWord = "PassWord";
        //设置实例的Vhost。
        String virtualHost = "test-vhost";

        Connection connection = createConnection(hostName, userName, passWord, virtualHost);
        final Channel channel = connection.createChannel();

        //声明Queue。
        String queueName = "test-queue";
        //创建${QueueName} ，该Queue必须在云消息队列RabbitMQ版控制台上已存在。
        AMQP.Queue.DeclareOk queueDeclareOk = channel.queueDeclare(queueName, true, false, false,
                new HashMap<String, Object>());

        //开始消费消息。
        channel.basicConsume(queueName, false, "test-consumerTag", new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理。
                System.out.println("Received： " + new String(body, StandardCharsets.UTF_8) + ", deliveryTag: "
                        + envelope.getDeliveryTag() + ", messageId: " + properties.getMessageId());
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

    public static Connection createConnection(String hostName, String userName, String passWord, String virtualHost)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(passWord);
        //设置为true,开启Connection自动恢复功能；设置为false,关闭Connection自动恢复功能。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        // 默认端口，非加密端口5672，加密端口5671。
        factory.setPort(5672);
        factory.setConnectionTimeout(300 * 1000);
        factory.setHandshakeTimeout(300 * 1000);
        factory.setShutdownTimeout(0);
        Connection connection = factory.newConnection();
        return connection;
    }
}

代码参数说明

1. 返回消息队列RabbitMQ版控制台页签，在左侧导航栏中，单击Dashboard。

2. 在服务关联角色对话框中，单击授权。

说明：如果您是首次使用Dashboard功能，需要进行授权。

3. 在开通ARMS服务对话框中，单击开通。

说明：如果您是首次使用Dashboard功能，需要开通ARMS服务。

4. 查看RabbitMQ实例的概览指标，可以查看每个Queue的消息堆积数，消息速率等指标。

5. 在左侧导航栏中，单击Queue列表。

6. 在Queue列表页面，单击test-queue。

7. 在test-queue详情页面，单击Dashboard。

8. 在Dashboard页签，可以查看指定Queue的详细指标变化趋势，用于定位问题。

9. 在左侧导航栏中，单击消息轨迹。

10. 在消息轨迹页面，您可以根据按Message ID查询和按Queue查询。

• 按Message ID查询：根据IntelliJ IDEA控制台打印的messageId可以精确查询对应消息的轨迹。

• 按Queue查询：根据Queue名称可以查询对应Queue下所有消息的轨迹。

11. 选择任意一条消息的轨迹，单击轨迹详情，查询对应消息的生产和投递轨迹详情。

1.为了验证RabbitMQ在高并发场景下的性能，我们可以通过多线程的方式来模拟大量消息的并发发送。具体代码如下：

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

public class ConcurrentProducerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 设置实例的接入点。
        String hostName = "xxx.xxx.aliyuncs.com";
        // 设置实例的静态用户名密码。
        String userName = "UserName";
        String passWord = "PassWord";
        // 设置实例的Vhost。
        String virtualHost = "test-vhost";

        // 创建连接
        Connection connection = createConnection(hostName, userName, passWord, virtualHost);
        Channel channel = connection.createChannel();

        // 设置Exchange、Queue和绑定关系。
        String exchangeName = "test-exchange";
        String queueName = "test-queue";
        String bindingKey = "test-routing-key";
        String exchangeType = "direct";

        // 创建线程池
        int threadCount = 50; // 线程数量
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

        // 开始发送消息
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    for (int j = 0; j < 1000; j++) { // 每个线程发送1000条
                        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(UUID.randomUUID().toString()).build();
                        channel.basicPublish(exchangeName, bindingKey, true, props,
                                ("并发消息发送示例Body-" + Thread.currentThread().getId() + "-" + j).getBytes(StandardCharsets.UTF_8));
                        System.out.println("[SendResult] Thread ID: " + Thread.currentThread().getId()
                                + ", Message sent successfully, messageId: " + props.getMessageId());
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            // 等待所有线程完成
        }

        // 关闭通道和连接
        channel.close();
        connection.close();
    }

    // 创建连接的方法
    public static Connection createConnection(String hostName, String userName, String passWord, String virtualHost)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(passWord);
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        factory.setPort(5672);
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);
        factory.setShutdownTimeout(0);
        return factory.newConnection();
    }
}

以上代码模拟50个用户并发发送消息的场景，总共发送5万条消息。

2.运行上述的并发生产者代码后，消费者代码依然使用之前的示例代码。参考之前的步骤，查看队列的Dashboard。

可以看到，当前队列的消息处理耗时为22ms，消息堆积在4分钟内完成发送，消息处理延时则最大为37ms。这表明，RabbitMQ能够在高并发场景下处理大量并发消息的发送和接收，保持高效的消息传递。低延时的表现确保了实时性要求较高的应用场景能够正常运行。