RabbitMQ 学习（四）-- 交换机（发布/订阅、路由、主题模式）

交换机

发布订阅模式： RabbitMQ tutorial - Publish/Subscribe — RabbitMQ

路由模式：RabbitMQ tutorial - Routing — RabbitMQ

主题模式：RabbitMQ tutorial - Topics — RabbitMQ

前面的学习每个任务都是都恰好交付给一个消费者(工作进程)。我们将消息传达给多个消费者。这种模式称为 “发布/订阅"。

为了说明这种模式，我们将构建一个简单的日志系统。它将由两个程序组成:第一个程序将发出日志消息，第二个程序是消费者。其中我们会启动两个消费者，其中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把消息打印在屏幕上，事实上第一个程序发出的日志消息将广播给所有消费者者。

1、Exchange（交换机）

1）概念

Exchange（交换机） message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到 Queue 中去。

RabbitMQ消息传递模型的核心思想是: **生产者生产的消息从不会直接发送到队列。**实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。

相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。

2）类型

Default Exchange（默认交换机） 、Direct Exchange（直接交换机） 、Fanout Exchange（扇出交换机） 、 Topic Exchange（主题交换机） 、 Headers Exchange（标题交换机）

系统中默认存在的交换机：

3）Default Exchange（默认交换机）

默认交换机是代理预先声明的没有名称（空字符串：""）的直接交换。 它有一个特殊的属性，使它对简单的应用程序非常有用：创建的每个队列都会自动绑定到它，并使用与队列名称相同的路由键。

例如，当您声明一个名为 “search-indexing-online” 的队列时，将使用 “search-indexing-online” 作为路由键（绑定键）将其绑定到默认交换机。 因此，使用路由键 “search-indexing-online” 发布到默认交换机的消息将被路由到队列 “search-indexing-online” 中。

默认的交换使看起来可以直接将消息传递到队列，但其实也通过了交换机。

channel.basicPublish("", QUEUE_NAME, null, message.getBytes());		// 默认交换机
1

第一个参数是交换机的名称。空字符串表示默认或无名称交换机

消息能路由发送到队列中其实是由 routingKey(bindingkey) 绑定 key 指定的，如果它存在的话。

2、临时队列

之前我们使用的是具有特定名称的队列( hello 和 ack_queue )。队列的名称我们来说至关重要：我们需要指定我们的消费者去消费哪个队列的消息。

每当我们连接到 RabbitMQ 时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接，队列将被自动删除。

创建方式：

String queueName = channel.queueDeclare().getQueue();
1

3、绑定（bindings）

绑定是交换机使用（除其他外）将消息路由到队列的规则。路由键就像一个过滤器。

它告诉我们 exchange 和 哪些队列进行了绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定：

4、Fanout Exchange（扇出交换机）

1）Fanout 概念

扇出交换机将消息路由绑定到它的所有队列，并且忽略路由键。 如果 “N” 个队列绑定到一个扇出交换机，当一条新消息发布到该交换机时，该消息的副本将被传递到所有 “N” 个队列。 扇出交换机是消息广播路由的理想选择。

扇出交换机可以用图形表示如下：

2）Fanout 实战测试

注意： fanout 模式中 routingKey 会被忽略

消费者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月01日 23:21
 */
public class ReceiveLog01 {
    // 交换机的名字
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args){
        new Thread(ReceiveLog01::subscribePrint, "subscribePrint").start();
        new Thread(ReceiveLog01::subscribeSaveFile, "subscribeSaveFile").start();
    }

    /**
     * @desc 接收到消息将消息打印到控制台
     * @auth llp
     * @date 2022/8/1 23:46 
     */
    private static void subscribePrint(){
        try {
            String threadName = "[" + Thread.currentThread().getName() + "]\t";
            Channel channel = RabbitMQUtil.getChannel();
            // 声明一个交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明一个队列 临时队列
            // 队列的名称是随机的，当与队列连接断开后，队列就自动删除
            String queueName = channel.queueDeclare().getQueue();
            // 绑定交换机和队列
            // routingkey 是为了找到对应关系的队列，但是这是广播交换机所以不需要专门找特定的队列，
            // 所以这个交换机忽略 routingkey 了，即这里写上也不会起作用。
            channel.queueBind(queueName, EXCHANGE_NAME, "");
            System.out.println(threadName + "等待接收消息...");
            // 接收消息
            // 消费成功的回调  consumerTag：与消费者关联的消费者标签
            DeliverCallback deliverCallback = (consumerTag, message) ->
                    System.out.println(threadName + "接收到的消息为 ==> " + new String(message.getBody(), StandardCharsets.UTF_8));
            // 消息消费失败的回调
            CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
            channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
        } catch (IOException | TimeoutException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * @desc 接收到消息将消息存到文件
     * @auth llp
     * @date 2022/8/1 23:47
     */
    private static void subscribeSaveFile() {
        try {
            String threadName = "[" + Thread.currentThread().getName() + "]\t";
            Channel channel = RabbitMQUtil.getChannel();
            // 声明一个交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明一个队列 临时队列
            // 队列的名称是随机的，当与队列连接断开后，队列就自动删除
            String queueName = channel.queueDeclare().getQueue();
            // 绑定交换机和队列
            // routingkey 是为了找到对应关系的队列，但是这是广播交换机所以不需要专门找特定的队列，
            // 所以这个交换机忽略 routingkey 了，即这里写上也不会起作用。
            channel.queueBind(queueName, EXCHANGE_NAME, "");
            System.out.println(threadName + "等待接收消息...");
            // 接收消息
            // 消费成功的回调  consumerTag：与消费者关联的消费者标签
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                Path path = Paths.get("D:\\IDEA\\RabbitMQ\\rabbitmq_log.txt");
                if (!Files.exists(path)){
                    Files.createFile(path);
                }
                try (FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.APPEND);){
                    ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());
                    int write = fileChannel.write(byteBuffer);
                    if (write > 0) {
                        System.out.println(threadName + "写入文件成功");
                    }else {
                        System.out.println(threadName + "写入文件失败");
                    }
                }catch (IOException e){
                    e.printStackTrace();
                }

            };
            // 消息消费失败的回调
            CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
            channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
        } catch (IOException | TimeoutException e) {
            throw new RuntimeException(e);
        }
    }
}
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

生产者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月01日 23:52
 */
public class EmitLog {
    // 交换机的名字
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 声明交换机
        // 也可以不声明，这样不用区分先启动消费者还是先启动生产者
        // 生产者中声明交换机和队列,那要用到的所有队列都需要在生产者创建并与交换机绑定；
        // 而在消费者中,只需要创建自己的队列并与交换机绑定
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

        // 发送消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()){
            String msg = scanner.next();
            // fanout 模式中 routingKey 会被忽略
            channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发出消息 ==> " + msg);
        }
    }
}
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

运行测试

5、Direct Exchange（直接交换机）

1）Direct 概念

直接交换机根据消息路由键将消息传递到队列。

直接交换机是消息的单播路由的理想选择（尽管它们也可用于多播路由）。 下面是它的工作原理：

• 队列使用路由键 “K” 绑定到交换机。

• 当带有路由键 “R” 的新消息到达直接交换机时，如果 “K = R”，交换机将其路由到队列。

直接交换机通常用于以循环方式在多个工作人员（同一应用程序的实例）之间分配任务。消息在消费者之间而不是队列之间进行负载平衡。

当绑定的 路由键 都相同时，就和 Fanout 模式一样了。

直接交换可以用图形表示如下：

2）Direct 实战

消费者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月02日 22:34
 */
public class ReceiveLogsDirect {
    // 交换机的名称
    private static final String DIRECT_EXCHANGE_NAME = "direct_log";
    // 队列的名称
    private static final String QUEUE_CONSOLE_NAME = "console";
    private static final String QUEUE_DISK_NAME = "disk";

    public static void main(String[] args){
        new Thread(()->{receiveLogsDirect(QUEUE_CONSOLE_NAME, new String[]{"info", "warning"});}, "ReceiveLogsDirect01").start();
        new Thread(()->{receiveLogsDirect(QUEUE_DISK_NAME, new String[]{"debug"});}, "ReceiveLogsDirect02").start();
    }

    private static void receiveLogsDirect(String queueName, String[] routingKeys){
        try {
            String threadName = "[" + Thread.currentThread().getName() + "]\t";
            Channel channel = RabbitMQUtil.getChannel();
            // 声明一个交换机
            channel.exchangeDeclare(DIRECT_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            // 声明队列
            channel.queueDeclare(queueName, false, false, false, null);
            for (String routingKey : routingKeys) {
                channel.queueBind(queueName, DIRECT_EXCHANGE_NAME, routingKey);
            }
            // 接收消息
            // 消费成功的回调  consumerTag：与消费者关联的消费者标签
            DeliverCallback deliverCallback = (consumerTag, message) ->
                    System.out.println(threadName + "接收到的消息为 ==> " + new String(message.getBody(), StandardCharsets.UTF_8));
            // 消息消费失败的回调
            CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
            channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
        } catch (IOException | TimeoutException e) {
            throw new RuntimeException(e);
        }
    }
}
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
34
35
36
37
38
39
40

生产者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月02日 23:48
 */
public class DirectLog {
    // 交换机的名字
    private static final String EXCHANGE_NAME = "direct_log";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        // 发送消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()){
            String inStr = scanner.next();
            String[] split = inStr.split(":");
            // fanout 模式中 routingKey 会被忽略
            channel.basicPublish(EXCHANGE_NAME, split[0], null, split[1].getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发出消息 ==> " + split[1]);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

测试

6、Topic Exchange（主题交换机）

1）Topic 概念

主题交换机基于消息路由键与用于将队列绑定到交换机的模式之间的匹配将消息路由到一个或多个队列。主题交换类型通常用于实现各种发布/订阅模式变体。主题交换通常用于消息的多播路由。

主题交换有非常广泛的用例。每当一个问题涉及多个消费者/应用程序选择性地选择他们想要接收的消息类型时，就应该考虑使用主题交换。

示例用途：

• 分发与特定地理位置相关的数据，例如销售点

• 由多个工作人员完成的后台任务处理，每个工作人员都能够处理特定的任务集

• 股票价格更新（以及其他类型财务数据的更新）

• 涉及分类或标记的新闻更新（例如，仅针对特定运动或团队）

• 云服务中各种服务的编排

• 分布式架构/特定于操作系统的软件构建或打包，其中每个构建器只能处理一个架构或操作系统

2）Topic 路由键

发送到类型是 topic 交换机的消息的 routingKey 不能随意写，它必须是一个单词列表，以点号分隔开。这些单词可以是任意单词，比如说: “stock.usd.nyse” , “nyse 1mw”,“quick.orange.rabbit” 这种类型的。当然这个单词列表最多不能超过255个字节。

主题交换功能强大，可以像其他交换机一样运行。

当队列与 #（hash）绑定键绑定时 - 无论路由键如何，它都会接收所有消息 - 就像使用 fanout 交换机一样。

当绑定中不使用特殊字符 *（star）和 #（hash）时，主题交换机的作用就像 direct交换机一样。

3）Topic 实战

消费者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月03日 23:00
 */
public class ReceiveLogsTopic {
    // 交换机的名称
    private static final String TOPIC_EXCHANGE_NAME = "topic_log";
    // 队列的名称
    private static final String QUEUE_Q1_NAME = "Q1";
    private static final String QUEUE_Q2_NAME = "Q2";
    public static void main(String[] args) {
        new Thread(()->{receiveLogsTopic(QUEUE_Q1_NAME, new String[]{"*.orange.*"});}, "Q1").start();
        new Thread(()->{receiveLogsTopic(QUEUE_Q2_NAME, new String[]{"*.*.rabbit", "lazy.#"});}, "Q2").start();
    }

    private static void receiveLogsTopic(String queueName, String[] routingKeys){
        try {
            String threadName = "[" + Thread.currentThread().getName() + "]\t";
            Channel channel = RabbitMQUtil.getChannel();
            // 声明一个交换机
            channel.exchangeDeclare(TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            // 声明队列
            // 写在生产者那边，第一次启动消费者就会报错，找不到队列。
            // 如果队列 autoDelete 为 false，启动完生产者之后，再重新启动消费者就不会报找不到队列错误
            channel.queueDeclare(queueName, false, false, false, null);
            for (String routingKey : routingKeys) {
                channel.queueBind(queueName, TOPIC_EXCHANGE_NAME, routingKey);
            }
            // 接收消息
            // 消费成功的回调  consumerTag：与消费者关联的消费者标签
            DeliverCallback deliverCallback = (consumerTag, message) ->
                    System.out.println(threadName + "接收到的消息为 ==> " + new String(message.getBody(), StandardCharsets.UTF_8));
            // 消息消费失败的回调
            CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
            channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
        } catch (IOException | TimeoutException e) {
            throw new RuntimeException(e);
        }
    }
}
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
34
35
36
37
38
39
40
41

生产者代码：

/**
 * @desc
 * @auth llp
 * @date 2022年08月03日 23:13
 */
public class TopicLog {
    // 交换机的名字
    private static final String EXCHANGE_NAME = "topic_log";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        // 发送消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()){
            String inStr = scanner.next();
            String[] split = inStr.split("=>");
            // fanout 模式中 routingKey 会被忽略
            channel.basicPublish(EXCHANGE_NAME, split[0], null, split[1].getBytes(StandardCharsets.UTF_8));
            System.out.println("routingKey: "+ split[0] + " 生产者发出消息: " + split[1]);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

运行测试