RabbitMQ通配符模式

原创

会洗碗的CV工程师

发布于 2024-04-29 17:13:36

3340

发布于 2024-04-29 17:13:36

文章被收录于专栏：消息中间件

一、概念

RabbitMQ通配符模式，也被称为主题模式（Topic Pattern），是一种消息传递模式，它允许消息生产者将消息发送到一个交换机（exchange），并使用通配符形式的路由键来描述消息的特性。消费者则可以使用通配符匹配来订阅感兴趣的消息。

在RabbitMQ通配符模式中，路由键由生产者定义，用于标识消息的目的地；而绑定则由消费者定义，用于指定消息的接收规则。这种模式通常使用通配符符号（和#）来匹配路由键。其中，“#”表示匹配零个或多个单词，“”表示匹配不多不少一个单词。

通配符模式的一个重要概念是通配符表达式（binding pattern），它定义了绑定的模式规则。通配符表达式可以包含多个单词和通配符符号，以及点号（.）进行分隔。通配符表达式越精确，匹配的消息越准确。

例如，假设路由键通常由一个或多个单词组成，多个单词之间以“.”分割。那么，通配符表达式“item.#”能够匹配如“item.insert.abc”或“item.insert”等路由键，而“item.*”则只能匹配“item.insert”这一层的路由键。

总的来说，RabbitMQ通配符模式通过匹配路由键和绑定模式来实现精确的消息过滤和匹配，从而实现灵活的消息路由和过滤。这种模式在需要根据消息的特定属性进行路由和过滤的场景中非常有用。

要使用RabbitMQ通配符模式，首先需要安装和配置RabbitMQ服务器，然后使用RabbitMQ的客户端库（如Java客户端库）来编写生产者和消费者代码。在生产者代码中，需要定义并发送带有路由键的消息；在消费者代码中，则需要定义绑定模式并订阅感兴趣的消息。

通配符模式(Topic)是在路由模式的基础上，给队列绑定带通配符的路由关键字，只要消息的RoutingKey能实现通配符匹配，就会将消息转发到该队列。通配符模式比路由模式更灵活，使用topic交换机。

通配符规则：

消息设置RoutingKey时，RoutingKey由多个单词构成，中间以 . 分割。
队列设置RoutingKey时， # 可以匹配任意多个单词， * 可以匹配任意一个单词。

二、编写生产者

这里编写生产者还是按照之前的来，一个是得到求爱信息，然后我做了对不起的事情，需要道歉，发送道歉消息；最后得到了他的拒绝，并向我提出做普通朋友。并且给这三个消息分别确定主题，代码如下：

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 2.建立连接
        Connection connection = connectionFactory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();

        /**
         * 4.创建交换机
         * 参数1：交换机名
         * 参数2：交换机类型
         * 参数3：交换机持久化
         */
        channel.exchangeDeclare("exchange_topic", BuiltinExchangeType.TOPIC, true);

        /**
         * 5.创建队列，如果队列已存在，则使用该队列
         * 参数1：队列名
         * 参数2：是否持久化，true表示MQ重启后队列还在。
         * 参数3：是否私有化，false表示所有消费者都可以访问，true表示只有第一次拥有它的消费者才能访问
         * 参数4：是否自动删除，true表示不再使用队列时自动删除队列
         * 参数5：其他额外参数
         */
        channel.queueDeclare("GET_LOVE", true, false, false, null);
        channel.queueDeclare("SEND_SORRY", true, false, false, null);
        channel.queueDeclare("GET_REFUSE", true, false, false, null);

        /**
         * 6.绑定队列
         * 参数1：队列名
         * 参数2：交换机名
         * 参数3：路由关键字，发布订阅模式写""即可
         */
        channel.queueBind("GET_LOVE", "exchange_topic", "#.love.#");
        channel.queueBind("SEND_SORRY", "exchange_topic", "#.sorry.#");
        channel.queueBind("GET_REFUSE", "exchange_topic", "#.refuse.#");

        // 7.发送消息
        /**
         * 参数1：交换机名，""表示默认交换机
         * 参数2：路由键，简单模式就是队列名
         * 参数3：其他额外参数
         * 参数4：要传递的消息字节数组
         */
        channel.basicPublish("exchange_topic", "love", null, ("哇靠，好想捏你脸了").getBytes());
        channel.basicPublish("exchange_topic", "sorry.refuse", null, ("对不起，我错了;互删吧咯").getBytes());
        channel.basicPublish("exchange_topic", "refuse", null, ("不不不，我们还是普通朋友").getBytes());

        // 6.关闭信道和连接
        channel.close();
        connection.close();
        System.out.println("--- 发送成功 ---");
    }
}

三、编写消费者

这里消费者获取消息的代码和之前差不多，这里也是放出第一个求爱消息的消费者代码如下：

public class Consumer_love {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 2.建立连接
        Connection connection = connectionFactory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();
        /**
         * 4.监听队列
         * 参数1：监听的队列名
         * 参数2：是否自动签收，如果设置为false，则需要手动确认消息已收到，否则MQ会一直发送消息
         * 参数3：Consumer的实现类，重写该类方法表示接受到消息后如何消费
         */
        channel.basicConsume("GET_LOVE",true, new DefaultConsumer(channel){
           @Override
           public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws UnsupportedEncodingException {
               String message = new String(body,"UTF-8");
               System.out.println("接受求爱，消息为: "+message);
           }
        });

    }
}

测试结果应当是收到一条求爱消息，然后发送对不起的消息，并且收到两条拒绝消息：