RabbitMQ—简单队列模式

阅读完需：约 4 分钟

在官网的教程中，描述了如上六类工作队列模式：

1. 简单队列模式：最简单的工作队列，其中一个消息生产者，一个消息消费者，一个队列。也称为点对点模式
2. 工作模式：一个消息生产者，一个交换器，一个消息队列，多个消费者。同样也称为点对点模式
3. 发布/订阅模式：无选择接收消息，一个消息生产者，一个交换器，多个消息队列，多个消费者。称为发布/订阅模式
4. 路由模式：在发布/订阅模式的基础上，有选择的接收消息，也就是通过 routing 路由进行匹配条件是否满足接收消息。
5. 主题模式：同样是在发布/订阅模式的基础上，根据主题匹配进行筛选是否接收消息，比第四类更灵活。
6. RPC模式：与上面其他5种所不同之处，类模式是拥有请求/回复的。也就是有响应的，上面5种都没有。

文章目录

• 1 实现功能
• 2 构建项目
  • 2.1 导入依赖
  • 2.2 封装Connection
• 3 生产者
• 4 消费者
• 5 测试环节
  • 5.1 启动RabbitMQ
  • 5.2 创建vhost
  • 5.3 依次运行Producer\Customer
• 6 简单队列总结

1 实现功能

功能描述：一个生产者 P 发送消息到队列 Q，一个消费者 C 接收

P 表示为生产者 、C 表示为消费者，红色表示队列。

2 构建项目

导入依赖

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

封装Connection

common模块中封装 rabbitmq 连接类

public class MQConnectionUtils {

    public static Connection newConnection() throws IOException, TimeoutException {
        /** 1.定义连接工厂 */
        ConnectionFactory factory = new ConnectionFactory();
         /** 2.设置服务器地址 */
        factory.setHost("127.0.0.1");
         /** 3.设置协议端口号 */
        factory.setPort(5672);
         /** 4.设置vhost */
        factory.setVirtualHost("test001_host");
         /** 5.设置用户名称 */
        factory.setUsername("guest");
         /** 6.设置用户密码 */
        factory.setPassword("guest");
         /** 7.创建新的连接 */
        Connection newConnection = factory.newConnection();
        return newConnection;
    }

}

3 生产者

生产者负责创建消息队列并发送消息入列，简单分为5步：

1. 获取连接
2. 创建通道
3. 创建队列声明
4. 发送消息
5. 关闭队列

public class Producer {

    /** 队列名称 */
    private static final String QUEUE_NAME = "test_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        /** 1.获取连接 */
        Connection newConnection = MQConnectionUtils.newConnection();
         /** 2.创建通道 */
        Channel channel = newConnection.createChannel();
         /** 3.创建队列声明 */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String msg = "我是生产者生成的消息";
        System.out.println("生产者发送消息:" + msg);
         /** 4.发送消息 */
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        channel.close();
        newConnection.close();
    }
}

4 消费者

消费者实现和生产者实现过程差不多，但是没有关闭连接和通道，是因为要消费者一直等待随时可能发来的消息，大致分为如下3步：

1. 获取连接
2. 获取通道
3. 监听队列

public class Customer {

    /** 队列名称 */
    private static final String QUEUE_NAME = "test_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        System.out.println("002");
        /** 1.获取连接 */
        Connection newConnection = MQConnectionUtils.newConnection();
         /** 2.获取通道 */
        Channel channel = newConnection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String msgString = new String(body, "UTF-8");
                System.out.println("消费者获取消息:" + msgString);
            }
        };
         /** 3.监听队列 */
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);

    }

}

5 测试环节

5.1 启动RabbitMQ

5.2 创建vhost

5.3 依次运行Producer\Customer

Producer生产者启动

Rabbit管理平台，三条消息

Customer消费者启动

6 简单队列总结

简单队列也称为点对点，即一个生产者对应一个消费者，生产者发送消息到队列，消费者在队列中取出消息消费。

生产者大致步骤：

1. 获取连接
2. 创建通道
3. 创建队列声明
4. 发送消息
5. 关闭队列

消费者大致步骤：

1. 获取连接
2. 获取通道
3. 监听队列
4. 消费消息