RabbitMQ 学习教程（八）主题模式 Topic_rabbitmq topic 实现

主题模式

direct 类型的交换机路由规则是完全匹配 BindingKey 和 RoutingKey，但是这种严格的匹配方式在很多情况下不能满足实际业务的需求。

topic 类型的交换机在匹配规则上进行了扩展，它与 direct 类型的交换机类似，也是将消息路由到 BindingKey 和 RoutingKey 相匹配的队列中，但这里的匹配规则有些不同，它约定：

1. RoutingKey 为一个点号 “.” 分割的字符串（被点号 “.” 分割开的每一段独立的字符串成为一个单词）。如：com.rabbitmq.client、java.util.concurrent
2. BindingKey 和 RountingKey 一样，也是点号 “.” 分割的字符串
3. BindingKey 中可以存在两种特殊的字符串 “*” 和 “#”，用于做模糊匹配。其中，* 用于匹配一个单词，# 用于匹配多个单词（可以是零个）。

模糊匹配举例：

1. com.#：可以匹配：com.zzc.java、com.zzc 等
2. com.*：可以匹配：com.zzc、com.java 等

Topic 模型如下图：

对上图进行举例说明：

1. 路由键为 com.rabbitmq.client 的消息会同时路由到 Q1、Q2、Q3
2. 路由键为 com.hidden.client 的消息会路由到 Q2、Q3
3. 路由键为 com.hidden.demo 的消息会路由到 Q3
4. 路由键为 java.util.concurrent 的消息会被丢弃或者返回给生产者，因为，它没有匹配任何路由键

后台页面管理

Topic 在 direct 模式上面进一步筛选

新建3个队列：topic_queue1、topic_queue2、topic_queue3

新建一个交换机 topic_exchange

绑定交换机和队列的关系（需要添加绑定键。# 代表0个或多个；* 代表一个）

在交换机中发送消息（需要指定路由键）

只有路由键和绑定键相匹配的队列才收到消息

代码实现

生产者
和路由模式不同：

1. 路由键 RoutingKey 由多个单词构成

public class Producer {

    private static final String TOPIC_EXCHANGE_NAME = "code_topic_exchange";

    public static void main(String[] args) throws Exception{
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");

        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        // 3. 通过通道声明交换机，以及交换机类型为 direct
        /**
         * @param1：交换机名称
         * @param2：交换机类型
         */
        channel.exchangeDeclare(TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

        // 4. 消息内容
        String message = "Hello RabbitMQ Topic!!";
        String routingKey = "com.rabbitmq.client";
        // 5. 发送消息到交换机，并指定路由键 RoutingKey 为 com.rabbitmq.client
        channel.basicPublish(TOPIC_EXCHANGE_NAME, routingKey, null, message.getBytes());
        System.out.println("消息发送完成~~~发送的消息为：" + message);
        // 6. 关闭信道、连接
        RabbitMqUtil.close(connection, channel);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

消费者
和路由模式不同：

1. 绑定键 BindingKey 中带有模糊匹配

public class Consumer {

    private static final String TOPIC_QUEUE_NAME = "topic_queue1";
    private static final String TOPIC_EXCHANGE_NAME = "code_topic_exchange";

    public static void main(String[] args) throws Exception{
        // 获取连接
        Connection connection = RabbitMqUtil.getConnection("消费者");
        // 获取通道
        Channel channel = connection.createChannel();
        String bindingKey = "*.rabbitmq.*";
        // 绑定队列到交换机，并指定一个绑定键 BindingKey
        channel.queueBind(TOPIC_QUEUE_NAME, TOPIC_EXCHANGE_NAME, bindingKey);

        // 定义消费者
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 消息体
                String msg = new String(body,"utf-8");
                System.out.println("收到消息：" + msg);
            }
        };

        // 监听队列
        channel.basicConsume(TOPIC_QUEUE_NAME, true, consumer);

        System.out.println("开始接收消息~~~");
        System.in.read();
        // 关闭信道、连接
        RabbitMqUtil.close(connection, channel);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

public class Consumer2 {

    private static final String TOPIC_QUEUE_NAME = "topic_queue2";
    private static final String TOPIC_EXCHANGE_NAME = "code_topic_exchange";

    public static void main(String[] args) throws Exception{
        // 获取连接
        Connection connection = RabbitMqUtil.getConnection("消费者");
        // 获取通道
        Channel channel = connection.createChannel();
        String bindingKey = "*.*.client";
        // 绑定队列到交换机，并指定一个绑定键 BindingKey
        channel.queueBind(TOPIC_QUEUE_NAME, TOPIC_EXCHANGE_NAME, bindingKey);

        // 定义消费者
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 消息体
                String msg = new String(body,"utf-8");
                System.out.println("收到消息：" + msg);
            }
        };

        // 监听队列
        channel.basicConsume(TOPIC_QUEUE_NAME, true, consumer);

        System.out.println("开始接收消息~~~");
        System.in.read();
        // 关闭信道、连接
        RabbitMqUtil.close(connection, channel);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33