5. RabbitMQ 消息队列中 Exchanges(交换机) 的具体说明

5. RabbitMQ 消息队列中 Exchanges(交换机) 的具体说明

@
目次

• 5. RabbitMQ 消息队列中 Exchanges(交换机) 的具体说明
  
• 1. Exchanges 交换机的概念
  
  
  
  • 1.1 Exchanges 的类型：
    
  
  
• 2. 临时队列
  
• 3. 绑定(bindings)
  
• 4. 扇出(fanout) 类型交换机
  
• 5. Direct(直接) exchange
  
• 6. Topic(主题) Exchange
  
• 7. 最后：
  

在上一节中，我们创建了一个工作队列。我们假设的是工作队列背后，每个任务都恰恰交付给一个消 费者(工作进程)。在这一部分中，我们将做一些完全不同的事情-我们将消息传达给多个消费者。这种模式 称为 ”发布/订阅”.
为了说明这种模式，我们将构建一个简朴的日志系统。它将由两个程序组成:第一个程序将发出日志消 息，第二个程序是消费者。此中我们会启动两个消费者，此中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把消息打印在屏幕上，究竟上第一个程序发出的日志消息将广播给所有消费者
1. Exchanges 交换机的概念

RabbitMQ 消息传递模拟的焦点思想是：**生产者生产的消息从不会直接发送到队列当中，而是通过发送给交换机(exchanges) ** 。现实上，通常生产者也不会知道将这些消息传递给了哪些队列中。
相反，**生产者只能将消息发送到交换机(exchange) ** ,交换机工作的内容非常简朴。

• 一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如那边理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是抛弃它们，这就的由交换机的类型来决定。
  

1.1 Exchanges 的类型：

总共有以下类型：

直接(direct), 主题(topic) ,标题(headers) , 扇出(fanout)

无名 exchange：
在前面，我们对 exchange 一无所知，但仍然可以或许将消息发送到队列。之前能实现的 原因是因为我们使用的是默认交换，我们通过空字符串(“”)举行标识。

一个参数是交换机的名称。空字符串表示默认或无名称交换机：消息能路由发送到队列中实在 是由 routingKey(bindingkey) 绑定 key 指定的，假如它存在的话。
2. 临时队列

我们使用的是具有特定名称的队列(还记得 hello 和 ack_queue 吗？)。队列的名称我们 来说至关紧张-我们需要指定我们的消费者去消费哪个队列的消息。
每当我们毗连到 Rabbit 时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称 的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连 接，队列将被自动删除。
创建临时队列的方式如下:

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

复制代码

创建出来之后长成这样:

3. 绑定(bindings)

什么是 bingding 呢，binding 实在是 exchange 和 queue 之间的桥梁，它告诉我们 exchange 和那个队 枚举行了绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 举行了绑定

4. 扇出(fanout) 类型交换机

Fanout 这种类型非常简朴。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。RabbitMQ 中默认有些 exchange 类型

Fanout 实战：

Logs 和临时队列的绑定关系如下图：

编写生产者，这里使用上交换机(fanout扇出(广播式的交换机))：将消息给 logs交换机发送消息

1. package com.rainbowsea.rabbitmq.five;
4. import com.rabbitmq.client.Channel;
5. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
7. import java.io.IOException;
8. import java.util.Scanner;
9. import java.util.concurrent.TimeoutException;
11. /**
12. * 发消息，交换机
13. */
14. public class EmitLog {
15.     private static final String EXCHANGE_NAME = "logs";
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         Channel channel = RabbitMQUtils.getChannel();
19.         // 声明将消息从哪个交换机当中发送出去
20.         channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
22.         Scanner scanner = new Scanner(System.in);
24.         while (scanner.hasNext()) {
25.             String message = scanner.next();
26.             channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8"));
27.             System.out.println("生产者发出的消息: " + message);
28.         }
29.     }
30. }

复制代码

编写两个消费者，读取 logs 交换机当中的信息。队列是随机生成的临时队列

1. package com.rainbowsea.rabbitmq.five;
3. import com.rabbitmq.client.Channel;
4. import com.rabbitmq.client.DeliverCallback;
5. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
7. import java.io.IOException;
8. import java.util.concurrent.TimeoutException;
10. public class ReceiveLogs01 {
12.     // 交换机名为 : logs
13.     private static final String EXCHANGE_NAME = "logs";
15.     public static void main(String[] args) throws IOException, TimeoutException {
16.         Channel channel = RabbitMQUtils.getChannel();
17.         // 声明一个交换机 fanout(扇出，广播式的)
18.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
20.         // 声明一个队列，临时队列
21.         /*
22.         生产一个临时队列，队列的名称是随机的
23.         当消费者断开与该队列的连接的时候，队列就自动删除
24.          */
25.         String queueName = channel.queueDeclare().getQueue();
27.         // 绑定交换机与队列
28.         channel.queueBind(queueName, EXCHANGE_NAME, "");
29.         System.out.println("等待接收消息，把接收到消息打印在屏幕上...");
31.         // 接收消息
32.         DeliverCallback deliverCallback = (consumerTag, message) -> {
33.             System.out.println("ReceiveLogs01 接收到的消息" + new String(message.getBody(), "UTF-8"));
36.         };
38.         // 消费者取消消息时回调接口
39.         channel.basicConsume(queueName, true, deliverCallback, consumerTage -> {
40.         });
41.     }
42. }

复制代码

1. package com.rainbowsea.rabbitmq.five;
3. import com.rabbitmq.client.Channel;
4. import com.rabbitmq.client.DeliverCallback;
5. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
7. import java.io.IOException;
8. import java.util.concurrent.TimeoutException;
10. public class ReceiveLogs02 {
12.     // 交换机名为 : logs
13.     private static final String EXCHANGE_NAME = "logs";
15.     public static void main(String[] args) throws IOException, TimeoutException {
16.         Channel channel = RabbitMQUtils.getChannel();
17.         // 声明一个交换机 fanout(扇出，广播式的)
18.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
20.         // 声明一个队列，临时队列
21.         /*
22.         生产一个临时队列，队列的名称是随机的
23.         当消费者断开与该队列的连接的时候，队列就自动删除
24.          */
25.         String queueName = channel.queueDeclare().getQueue();
27.         // 绑定交换机与队列
28.         channel.queueBind(queueName, EXCHANGE_NAME, "");
29.         System.out.println("等待接收消息，把接收到消息打印在屏幕上...");
31.         // 接收消息
32.         DeliverCallback deliverCallback = (consumerTag, message) -> {
33.             System.out.println("ReceiveLogs02接收到的消息" + new String(message.getBody(), "UTF-8"));
36.         };
38.         // 消费者取消消息时回调接口
39.         channel.basicConsume(queueName, true, deliverCallback, consumerTage -> {
40.         });
41.     }
42. }

复制代码

运行测试：效果

补充：
ReceiveLogs01 将接收到的消息打印在控制台

1. package com.rainbowsea.rabbitmq.five;
3. public class ReceiveLogs01 {
4.     private static final String EXCHANGE_NAME = "logs";
6.     public static void main(String[] argv) throws Exception {
7.         Channel channel = RabbitUtils.getChannel();
8.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
9. /**
10. * 生成一个临时的队列    队列的名称是随机的
11. * 当消费者断开和该队列的连接时  队列自动删除
12. */
13.         String queueName = channel.queueDeclare().getQueue();
14. //把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串 channel.queueBind(queueName, EXCHANGE_NAME, "");
15.         System.out.println("等待接收消息,把接收到的消息打印在屏幕.....");
16.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
17.             String message = new String(delivery.getBody(), "UTF-8");
18.             System.out.println("控制台打印接收到的消息" + message);
19.         };
20.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
21.         });
22.     }
23. }

复制代码

ReceiveLogs02 将接收到的消息存储在磁盘

1. package com.rainbowsea.rabbitmq.five;
3. public class ReceiveLogs02 {
4.     private static final String EXCHANGE_NAME = "logs";
6.     public static void main(String[] argv) throws Exception {
7.         Channel channel = RabbitUtils.getChannel();
8.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
9. /**
10. * 生成一个临时的队列    队列的名称是随机的
11. * 当消费者断开和该队列的连接时  队列自动删除
12. */
13.         String queueName = channel.queueDeclare().getQueue();
14. //把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串 channel.queueBind(queueName, EXCHANGE_NAME, "");
15.         System.out.println("等待接收消息,把接收到的消息写到文件.....");
16.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
17.             String message = new String(delivery.getBody(), "UTF-8");
18.             File file = new File("C:\\work\\rabbitmq_info.txt");
19.             FileUtils.writeStringToFile(file, message, "UTF-8");
20.             System.out.println("数据写入文件成功");
21.         };
22.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
23.         });
24.     }
25. }

复制代码

EmitLog 发送消息给两个消费者接收

1. package com.rainbowsea.rabbitmq.five;
3. public class EmitLog {
4.     private static final String EXCHANGE_NAME = "logs";
6.     public static void main(String[] argv) throws Exception {
7.         try (Channel channel = RabbitUtils.getChannel()) {
8. /**
9. * 声明一个 exchange
10. * 1.exchange 的名称
11. * 2.exchange 的类型
12. */
13.             channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
14.             Scanner sc = new Scanner(System.in);
15.             System.out.println("请输入信息");
16.             while (sc.hasNext()) {
17.                 String message = sc.nextLine();
18.                 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
19.                 System.out.println("生产者发出消息" + message);
20.             }
21.         }
22.     }
23. }

复制代码

5. Direct(直接) exchange

上面我们构建了一个简朴的日志纪录系统。我们可以或许向许多接收者广播日志消息。在本 节我们将向此中添加一些特别的功能-比方说我们只让某个消费者订阅发布的部分消息。例如我们只把 严峻错误消息定向存储到日志文件(以节流磁盘空间)，同时仍然可以或许在控制台上打印所有日志消息。
我们再次往返顾一下什么是 bindings，绑定是交换机和队列之间的桥梁关系。也可以这么理解： 队列只对它绑定的交换机的消息感兴趣。绑定用参数：routingKey 来表示也可称该参数为 binding key， 创建绑定我们用代码: channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");绑定之后的 意义由其交换类型决定。
我们的日志系统将所有消息广播给所有消费者，对此我们想做一些改变，例如我们希 望将日志消息写入磁盘的程序仅接收严峻错误(errros)，而不存储哪些告诫(warning)或信息(info)日志 消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性-它只能举行无意识的 广播，在这里我们将使用 direct 这种类型来举行更换，这种类型的工作方式是，消息只去到它绑定的 routingKey 队列中去。
根据路由键精确匹配（一模一样）举行路由消息队列；

• 在上面这张图中，我们可以看到 X 绑定了两个队列，绑定类型是 direct。队列 Q1 绑定键为 orange， 队列 Q2 绑定键有两个:一个绑定键为 black，另一个绑定键为 green.
  
• 在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 blackgreen 和的消息会被发布到队列 Q2，其他消息类型的消息将被抛弃。
  

实战：

编写生产者，向 (DIRECT 直接类型的交换机)，当中的 队列的 Routing key 为error 的队列发送消息。

1. package com.rainbowsea.rabbitmq.six;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
7. import java.io.IOException;
8. import java.util.Scanner;
9. import java.util.concurrent.TimeoutException;
11. public class DirectLogs {
13.     // 交换机名为 : direct_logs
14.     private static final String EXCHANGE_NAME = "direct_logs";
16.     public static void main(String[] args) throws IOException, TimeoutException {
17.         Channel channel = RabbitMQUtils.getChannel();
18.         // 声明一个交换机 BuiltinExchangeType.DIRECT(直接)
19.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
20.         Scanner scanner = new Scanner(System.in);
22.         while (scanner.hasNext()) {
23.             String message = scanner.next();
24.             channel.basicPublish(EXCHANGE_NAME,"error",null,message.getBytes("UTF-8"));
25.             System.out.println("生产者发出消息" + message);
26.         }
27.     }
28. }

复制代码

编写两个消费者：

• 一个消费者，仅仅读取：队列为 disk  并且Routing key 为 error 的消息内容
  

1. package com.rainbowsea.rabbitmq.six;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.DeliverCallback;
6. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
8. import java.io.IOException;
9. import java.util.concurrent.TimeoutException;
11. public class ReceiveLogsDirect01 {
13.     // 交换机名为 : direct_logs
14.     private static final String EXCHANGE_NAME = "direct_logs";
16.     public static void main(String[] args) throws IOException, TimeoutException {
17.         Channel channel = RabbitMQUtils.getChannel();
18.         // 声明一个交换机 BuiltinExchangeType.DIRECT(直接)
19.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
21.         // 声明一个队列,队列名为 disk
22.         channel.queueDeclare("disk", false, false, false, null);
23.         // 绑定交换机与队列, 该 disk 队列的 Routing key 为 error
24.         channel.queueBind("disk", EXCHANGE_NAME, "error");
26.         System.out.println("ReceiveLogsDirect01 等待接收消息，把接收到消息打印在屏幕上...");
28.         // 接收消息
29.         DeliverCallback deliverCallback = (consumerTag, message) -> {
30.             System.out.println("ReceiveLogsDirect01 接收到的消息" + new String(message.getBody(), "UTF-8"));
33.         };
35.         // 消费者取消消息时回调接口
36.         // 第一个参数，队列
37.         channel.basicConsume("disk", true, deliverCallback, consumerTage -> {
38.         });
39.     }
40. }

复制代码

• 一个消费者，仅仅读取：队列为 console  并且Routing key 为 info 或者是 warning 的消息内容
  

1. package com.rainbowsea.rabbitmq.six;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.DeliverCallback;
6. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
8. import java.io.IOException;
9. import java.util.concurrent.TimeoutException;
11. public class ReceiveLogsDirect02 {
13.     // 交换机名为 : direct_logs
14.     private static final String EXCHANGE_NAME = "direct_logs";
16.     public static void main(String[] args) throws IOException, TimeoutException {
17.         Channel channel = RabbitMQUtils.getChannel();
18.         // 声明一个交换机  BuiltinExchangeType.DIRECT(直接)
19.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
21.         // 声明一个队列,队列名为 disk
22.         channel.queueDeclare("console", false, false, false, null);
23.         // 绑定交换机与队列, 该 console 队列的 Routing key 为 info和 warning 两个
24.         channel.queueBind("console", EXCHANGE_NAME, "info");
25.         channel.queueBind("console", EXCHANGE_NAME, "warning");
27.         System.out.println("ReceiveLogsDirect02 等待接收消息，把接收到消息打印在屏幕上...");
29.         // 接收消息
30.         DeliverCallback deliverCallback = (consumerTag, message) -> {
31.             System.out.println("ReceiveLogsDirect02 接收到的消息" + new String(message.getBody(), "UTF-8"));
34.         };
36.         // 消费者取消消息时回调接口
37.         // 第一个参数，队列 console
38.         channel.basicConsume("console", true, deliverCallback, consumerTage -> {
39.         });
40.     }
41. }

复制代码

运行测试：

多重绑定：

固然假如 exchange 的绑定类型是 direct，但是它绑定的多个队列的 key 假如都相同，在这种情 况下固然绑定类型是 direct 但是它表现的就和 fanout 有点类似了，就跟广播差不多，如上图所示。

6. Topic(主题) Exchange

我们改进了日志纪录系统。我们没有使用只能举行随意广播的 fanout 交换机，而是 使用了 direct 交换机，从而有能实现有选择性地接收日志。
尽管使用 direct 交换机改进了我们的系统，但是它仍然存在局限性-比方说我们想接收的日志类型有 info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时间 direct 就办不到了。这个时间 就只能使用 topic 类型
Topic 的要求：
发送到类型是 topic 交换机的消息的routing key 不能随意写，必须满足一定的要求，它必须是一个单词列表，并且以点号. 分隔开 。这些单词可以是任意单词，比如说："stock.usd.nyse","nyse.vmw","quick.orange.rabbit" 这种类型的。固然这个单词列表最多不能超过 255 个字节。
在这个规则列表中，此中有两个更换符是特别需要注意的：

• *(星号) 可以代替一个单词，注意是一个单词，不是一个字母
  
• #(井号) 可以更换零个或多个单词，注意是一个单词，不是一个字母
  

当队列绑定关系是下列这种情况时需要引起注意

• 当一个队列绑定键是 # ，那么这个队列将接收所有数据，就有点像 fanout 扇出，广播
  
• 假如队列绑定键当中没有 # 和 * 出现，那么该队列绑定类型就是 direct 直接匹配的方式
  

Topic 匹配案例：
下图绑定关系如下

• Q1-->绑定的是
  
  
  
  • 中间带 orange 带 3 个单词的字符串 (*.orange.*)
    
  
  
• Q2-->绑定的是
  
  
  
  • 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)  第一个单词是 lazy 的多个单词(lazy.#)
    
  
  

上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的

1. quick.orange.rabbit         被队列 Q1Q2 接收到
2. lazy.orange.elephant        被队列 Q1Q2 接收到
3. quick.orange.fox            被队列 Q1 接收到
4. lazy.brown.fox              被队列 Q2 接收到
5. lazy.pink.rabbit            虽然满足两个绑定但只被队列 Q2 接收一次
6. quick.brown.fox             不匹配任何绑定不会被任何队列接收到会被丢弃
7. quick.orange.male.rabbit    是四个单词不匹配任何绑定会被丢弃
8. lazy.orange.male.rabbit     是四个单词但匹配 Q2

复制代码

编写生产者，向 topic_logs 交换机(TOPIC 主题的交换机)发送消息 ,这里我们使用上 Map 存储发送的信息。

1. package com.rainbowsea.rabbitmq.seven;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
7. import java.io.IOException;
8. import java.util.HashMap;
9. import java.util.Map;
10. import java.util.Scanner;
11. import java.util.concurrent.TimeoutException;
13. public class EmitLogTopic {
15.     // 交换机名为 : topic_logs
16.     private static final String EXCHANGE_NAME = "topic_logs";
18.     /*
19.     - Q1-->绑定的是
20.         - 中间带 orange 带 3 个单词的字符串 (*.orange.*)
21.    - Q2-->绑定的是
22.        - 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)  第一个单词是 lazy 的多个单词(lazy.#)
24.      */
26.     public static void main(String[] args) throws IOException, TimeoutException {
27.         Channel channel = RabbitMQUtils.getChannel();
28.         // 声明一个交换机 BuiltinExchangeType.TOPIC(主题)
29.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
31.         Map<String, String> bindingKeyMap = new HashMap<>();
32.         bindingKeyMap.put("quick.orange.rabbit", "被队列 Q1Q2 接收到");
33.         bindingKeyMap.put("lazy.orange.elephant", "被队列 Q1Q2 接收到");
34.         bindingKeyMap.put("quick.orange.fox", "被队列 Q1 接收到");
35.         bindingKeyMap.put("lazy.brown.fox", "被队列 Q2 接收到");
36.         bindingKeyMap.put("lazy.pink.rabbit", "虽然满足两个绑定但只被队列 Q2 接收一次");
37.         bindingKeyMap.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃");
38.         bindingKeyMap.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃");
39.         bindingKeyMap.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2");
41.         for (Map.Entry<String, String> bindingKeyEntry : bindingKeyMap.entrySet()) {
42.             String routingKey = bindingKeyEntry.getKey();
43.             String message = bindingKeyEntry.getValue();
44.             channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
45.             System.out.println("生产者发出消息: " + message);
47.         }
48.     }
49. }

复制代码

编写一个消费者，仅仅读取：Routing key 为 *.orange.* 匹配的队列的消息

1. package com.rainbowsea.rabbitmq.seven;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.DeliverCallback;
6. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
8. import java.io.IOException;
9. import java.util.concurrent.TimeoutException;
11. public class ReceiveLogsTopic01 {
13.     // 交换机名为 : topic_logs
14.     private static final String EXCHANGE_NAME = "topic_logs";
16.     public static void main(String[] args) throws IOException, TimeoutException {
17.         Channel channel = RabbitMQUtils.getChannel();
19.         // 声明一个交换机 BuiltinExchangeType.TOPIC(主题)
20.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
22.         // 声明一个队列,队列名为 Q1
23.         String queueName = "Q1";
24.         channel.queueDeclare(queueName, false, false, false, null);
25.         // 绑定交换机与队列, 该 Q1 队列的 Routing key 为 *.orange.*
26.         channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
28.         System.out.println("Q1 等待接收消息，把接收到消息打印在屏幕上...");
30.         // 接收消息
31.         DeliverCallback deliverCallback = (consumerTag, message) -> {
32.             System.out.println("Q1 接收到的消息" + new String(message.getBody(), "UTF-8"));
33.             System.out.println("Q1 接收队列:" + queueName + "绑定键" + message.getEnvelope().getRoutingKey());
36.         };
38.         // 消费者取消消息时回调接口
39.         // 第一个参数，队列 queueName
40.         channel.basicConsume(queueName, true, deliverCallback, consumerTage -> {
41.         });
42.     }
43. }

复制代码

编写一个消费者，仅仅读取：Routing key 为 *.orange.* 和 "lazy.#" 匹配的队列的消息

1. package com.rainbowsea.rabbitmq.seven;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.DeliverCallback;
6. import com.rainbowsea.rabbitmq.utils.RabbitMQUtils;
8. import java.io.IOException;
9. import java.util.concurrent.TimeoutException;
11. public class ReceiveLogsTopic02 {
13.     // 交换机名为 : topic_logs
14.     private static final String EXCHANGE_NAME = "topic_logs";
16.     public static void main(String[] args) throws IOException, TimeoutException {
17.         Channel channel = RabbitMQUtils.getChannel();
19.         // 声明一个交换机 BuiltinExchangeType.TOPIC(主题)
20.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
22.         // 声明一个队列,队列名为 Q2
23.         String queueName = "Q2";
24.         channel.queueDeclare(queueName, false, false, false, null);
25.         // 绑定交换机与队列, 该 Q2 队列的 Routing key 为 *.orange.* 和 "lazy.#"
26.         channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
27.         channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
29.         System.out.println("Q2 等待接收消息，把接收到消息打印在屏幕上...");
31.         // 接收消息
32.         DeliverCallback deliverCallback = (consumerTag, message) -> {
33.             System.out.println("Q2 接收到的消息" + new String(message.getBody(), "UTF-8"));
34.             System.out.println("Q2 接收队列:" + queueName + "绑定键" + message.getEnvelope().getRoutingKey());
37.         };
39.         // 消费者取消消息时回调接口
40.         // 第一个参数，队列 queueName
41.         channel.basicConsume(queueName, true, deliverCallback, consumerTage -> {
42.         });
43.     }
44. }

复制代码

运行执行效果：

从运行结果上看，我们可以显着的看到，消费者根据对应的 Routing key  匹配而读取，对应的消息队列上的消息内容。

7. 最后：

“在这个最后的篇章中，我要表达我对每一位读者的感激之情。你们的关注和复兴是我创作的动力源泉，我从你们身上吸取了无尽的灵感与勇气。我会将你们的鼓励留在心底，继续在其他的领域奋斗。感谢你们，我们总会在某个时刻再次相遇。”

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。