【RabbitMQ笔记05】消息队列RabbitMQ七种模式之Routing路由键模式

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

这篇文章主要介绍消息队列RabbitMQ七种模式之Routing路由键模式。

一、消息队列

一、消息队列

1.1、Routing模式

RabbitMQ中提供的Routing模式就是根据RoutingKey来决定这条消息应该分发到哪个Queue队列里面一般的Routing模式中Exchange交换机都是采用【direct】直接模式。生产者发送的每一条消息都会携带一个routing keyExchange根据这个routing key将其分发到对应的Queue队列里面。一个Queue队列可以绑定多个routing key。

注意如果Exchange交换机接收到的消息没法根据routing key将其分发到指定的Queue队列里面那么这一条消息默认就会被丢弃。

1.2、案例代码

1引入依赖

<!-- 引入 RabbitMQ 依赖 -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.16.0</version>
</dependency>

2编写生产者

生产者发送不同类型的Routing Key的消息。

package com.rabbitmq.demo.routing;

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

/**
 * @version 1.0.0
 * @Date: 2023/2/25 16:23
 * @Copyright (C) ZhuYouBin
 * @Description: 消息生产者
 */
public class Producer {
    public static void main(String[] args) {
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 2、设置连接的 RabbitMQ 服务地址
        factory.setHost("127.0.0.1"); // 默认就是本机
        factory.setPort(5672); // 默认就是 5672 端口
        // 3、获取连接
        Connection connection = null; // 连接
        Channel channel = null; // 通道
        try {
            connection = factory.newConnection();
            // 4、获取通道
            channel = connection.createChannel();
            // 5、声明 Exchange如果不存在则会创建
            String exchangeName = "exchange_direct_2023";
            channel.exchangeDeclare(exchangeName, "direct");
            // 6、发送消息
            for (int i = 0; i < 10; i++) {
                // 路由键唯一标识
                String routingKey = "error";
                if (i % 3 == 0) {
                    routingKey = "info";
                } else if (i % 3 == 1) {
                    routingKey = "warn";
                }
                String message = "这是Routing模式发送的第【" + (i+1) + "】条【" + routingKey + "】消息数据";
                channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != channel) {
                try {
                    channel.close();
                } catch (Exception e) {}
            }
            if (null != connection) {
                try {
                    connection.close();
                } catch (Exception e) {}
            }
        }
    }
}

3编写消费者

不同的消费者可以从同一个Queue队列里面获取不同的Routing Key的消息。
假设消费者A接收routingKey="info"的消息消费者B接收routingKey="error"的消息。

package com.rabbitmq.demo.routing;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @version 1.0.0
 * @Date: 2023/2/25 16:30
 * @Copyright (C) ZhuYouBin
 * @Description: 消息消费者
 */
public class Consumer {
    public static void main(String[] args) {
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 2、设置连接的 RabbitMQ 服务地址
        factory.setHost("127.0.0.1"); // 默认就是本机
        factory.setPort(5672); // 默认就是 5672 端口
        // 3、获取连接
        Connection connection = null; // 连接
        Channel channel = null; // 通道
        try {
            connection = factory.newConnection();
            // 4、获取通道
            channel = connection.createChannel();
            // 5、声明 Exchange如果不存在则会创建
            String exchangeName = "exchange_direct_2023";
            channel.exchangeDeclare(exchangeName, "direct");
            // 6、指定需要操作的消息队列如果队列不存在则会创建
            String queueName = "queue_direct_2023";
            channel.queueDeclare(queueName, false, false, false, null);
            // 7、绑定 Exchange 和 Queue, 接收 routingKey = "info" 的消息
            channel.queueBind(queueName, exchangeName, "info");
            // 8、消费消息
            DeliverCallback callback = new DeliverCallback() {
                public void handle(String s, Delivery delivery) throws IOException {
                    // 接收消息
                    System.out.println("这是接收的消息" + new String(delivery.getBody()));
                }
            };
            channel.basicConsume(queueName, true, callback, i->{});
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

运行消费者、生产者查看控制台日志输出结果。