RabbitMQ学习(七):交换器

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: RabbitMQ学习(七):交换器

〇、前言



在之前的内容中,我们创建了一个工作队列。我们假设的是工作队列背后,每个任务都恰好交付给一个消 费者(工作进程)。在今天的内容中,我们将做一些完全不同的事情——我们将消息传达给多个消费者。这种模式 称为 “发布/订阅”。

071e950a537787b9a863ebeb07122c32.png


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


一、Exchanges 概念



RabbitMQ 消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。实际上,通常生产者甚至都不知道这些消息传递传递到了哪些队列中。相反,生产者只能将消息发送到交换机(exchange)。


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


1.1 Exchanges 的类型


总共有以下类型: 直接(direct), 主题(topic) ,标题(headers) , 扇出(fanout)


1.2 无名 exchange


在前面部分中我们对 exchange 没有任何操作,但仍然能够将消息发送到队列。 原因是我们使用的是默认交换,我们通过空字符串(“”)进行标识。

channel.basicPublish("", "hello", null, message.getBytes());


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


1.3 绑定


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

0b81e13c0642e7e14d0b23f0cc1cd77b.png




二、fanout类型



2.1 fanout概述


Fanout 这种类型非常简单。正如名称,它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些 exchange类型。


2.2 fanout实战

294d3af076543666be3101c9e29ce7ea.png


消费者1将接受到的消息打印在控制台,核心代码:

//绑定fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//创建一个临时的队列 队列的名称是随机的,当消费者断开和该队列的连接时 队列自动删除
String queueName = channel.queueDeclare().getQueue();
//把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println("等待接收消息,把接收到的消息打印在屏幕.....");


消费者2将接受到的消息保存在磁盘,核心代码:

//绑定fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//生成一个临时的队列 队列的名称是随机的,当消费者断开和该队列的连接时 队列自动删除
String queueName = channel.queueDeclare().getQueue();
//把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
channel.queueBind(queueName, EXCHANGE_NAME, "");
//写入磁盘
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     File file = new File("C:\\work\\rabbitmq_info.txt");
     FileUtils.writeStringToFile(file,message,"UTF-8");
     System.out.println("数据写入文件成功");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });


生产者核心代码:

/**
 * 声明一个 exchange
 * 1.exchange 的名称
 * 2.exchange 的类型
 */
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//发消息
while (sc.hasNext()) {
     channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
     System.out.println("生产者发出消息" + message);
}


结果是两个消费者都能收到生产者发送的消息。


需要注意的是,交换机把消息分发给了两个队列,在每个队列中一个消息仍然只是被消费了一次。


三、direct类型



3.1 direct概述


首先我们再次来回顾一下什么是 bindings,绑定是交换机和队列之间的桥梁关系。也可以这么理解:队列只对它绑定的交换机的消息感兴趣。绑定用参数:routingKey 来表示也可称该参数为 binding key,创建绑定我们用代码:channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");绑定之后的 意义由其交换类型决定。


上一节中的我们的日志系统将所有消息广播给所有消费者,对此我们想做一些改变,例如我们希 望将日志消息写入磁盘的程序仅接收严重错误(errros),而不存储哪些警告(warning)或信息(info)日志 消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性——它只能进行无意识的 广播。在这里我们将使用 direct 这种类型来进行替换,这种类型的工作方式是,消息只去到它绑定 routingKey 队列中去。

37f749a266d07e7ebd1d5e6558663872.png

在上面这张图中,我们可以看到 X 绑定了两个队列,绑定类型是 direct。队列 Q1 绑定键为orange,队列 Q2 绑定键有两个:一个绑定键为 black,另一个绑定键为 green。


在这种绑定情况下,生产者发布消息到 exchange 上,绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 blac和kgreen 的消息会被发布到队列 Q2,其他消息类型的消息将被丢弃。

a8443a458090eadfc0ba7c0e4d7ad931.png


当然如果 exchange 的绑定类型是 direct,但是它绑定的多个队列的 key 如果都相同,在这种情 况下虽然绑定类型是 direct 但是它表现的就和 fanout 有点类似了,就跟广播差不多,如上图所示。这也被称为多重绑定。


3.2 direct实战

a4e8bccdcc7390f5f98acc092807c13d.png

消费者1(处理error)核心代码:

channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.queueDeclare("disk", false, false, false, null);
//绑定error
channel.queueBind(queueName, EXCHANGE_NAME, "error");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     message="接收绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message;
     FileUtils.writeStringToFile(new File("C:\\work\\rabbitmq_info.txt"), message, "UTF-8");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});


消费者2(处理info和warning)核心代码:

channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.queueDeclare("console", false, false, false, null);
//绑定info和warning
channel.queueBind(queueName, EXCHANGE_NAME, "info");
channel.queueBind(queueName, EXCHANGE_NAME, "warning");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     System.out.println("绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {})


生产者核心代码:

channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
//创建多个 bindingKey
Map<String, String> bindingKeyMap = new HashMap<>();
bindingKeyMap.put("info","普通 info 信息");
bindingKeyMap.put("warning","警告 warning 信息");
bindingKeyMap.put("error","错误 error 信息");
//debug 没有消费这接收这个消息 所有就丢失了
bindingKeyMap.put("debug","调试 debug 信息");
//发消息
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){
    String bindingKey = bindingKeyEntry.getKey();
    String message = bindingKeyEntry.getValue();
    channel.basicPublish(EXCHANGE_NAME,bindingKey, null,
    message.getBytes("UTF-8"));
}



四、topic类型



4.1 topic类型的作用

在上一个小节中,我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机,而是 使用了 direct 交换机,从而有能实现有选择性地接收日志。


尽管使用 direct 交换机改进了我们的系统,但是它仍然存在局限性——比方说我们想接收的日志类型有 info.base 和 info.advantage,某个队列只想 info.base 的消息,那这个时候 direct 就办不到了。这个时候 就只能使用 topic 类型。


4.2 topic类型的要求


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


在这个规则列表中,其中有两个替换符是大家需要注意的 :*(星号)可以代替一个单词 、#(井号)可以替代零个或多个单词。


4.3 topic匹配案例


下图绑定关系如下 :


Q1-->绑定的是 :中间带 orange 带 3 个单词的字符串(*.orange.*)

Q2-->绑定的是 :最后一个单词是 rabbit 的 3 个单词(*.*.rabbit) 、


第一个单词是 lazy 的多个单词(lazy.#)

1a7d89e2cf065e0cc6f4344532ed55f4.png


匹配结果如下表:

quick.orange.rabbit

被队列 Q1、Q2 接收到

lazy.orange.elephant

被队列 Q1、Q2 接收到

quick.orange.fox

被队列 Q1 接收到

lazy.brown.fox

被队列 Q2 接收到

lazy.pink.rabbit

虽然满足两个绑定,但只被队列 Q2 接收一次

quick.brown.fox

不匹配任何绑定,被丢弃

quick.orange.male.rabbit

是四个单词,不匹配任何绑定,被丢弃

lazy.orange.male.rabbit

是四个单词,但匹配 Q2


4.4 topic类型与fanout、direct的关系


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

  • 当一个队列绑定键是#,那么这个队列将接收所有数据,其实是fanout类型
  • 如果队列绑定键当中没有#和*出现,那么该队列绑定类型就是 direct 类型


4.5 topic实战


消费者1核心代码:

channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//声明 Q1 队列与绑定关系
channel.queueDeclare("Q1", false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println("接收队列:"+queueName+" 绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});


消费者2核心代码:

channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//声明 Q2 队列与绑定关系
channel.queueDeclare("Q2", false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
    System.out.println("接收队列:"+queueName+" 绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});


生产者核心代码:

//交换机
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// Q1-->绑定的是 中间带 orange 带 3 个单词的字符串(*.orange.*)
// Q2-->绑定的是 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)
//                 第一个单词是 lazy 的多个单词(lazy.#)
Map<String, String> bindingKeyMap = new HashMap<>();
bindingKeyMap.put("quick.orange.rabbit","被队列 Q1Q2 接收到");
bindingKeyMap.put("lazy.orange.elephant","被队列 Q1Q2 接收到");
bindingKeyMap.put("quick.orange.fox","被队列 Q1 接收到");
bindingKeyMap.put("lazy.brown.fox","被队列 Q2 接收到");
bindingKeyMap.put("lazy.pink.rabbit","虽然满足两个绑定但只被队列 Q2 接收一次");
bindingKeyMap.put("quick.brown.fox","不匹配任何绑定不会被任何队列接收到会被丢弃");
bindingKeyMap.put("quick.orange.male.rabbit","是四个单词不匹配任何绑定会被丢弃");
bindingKeyMap.put("lazy.orange.male.rabbit","是四个单词但匹配 Q2");
//发消息
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){
    String bindingKey = bindingKeyEntry.getKey();
    String message = bindingKeyEntry.getValue();
    channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes("UTF-8"));
}



相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
6月前
|
消息中间件 Java API
RocketMQ事务消息, 图文、源码学习探究~
介绍 RocketMQ是阿里巴巴开源的分布式消息中间件,它是一个高性能、低延迟、可靠的消息队列系统,用于在分布式系统中进行异步通信。 从4.3.0版本开始正式支持分布式事务消息~ RocketMq事务消息支持最终一致性:在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。 原理、流程 本质上RocketMq的事务能力是基于二阶段提交来实现的 在消息发送上,将二阶段提交与本地事务绑定 本地事务执行成功,则事务消息成功,可以交由Consumer消费 本地事务执行失败,则事务消息失败,Consumer无法消费 但是,RocketMq只能保证本地事务
|
6月前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
153 0
|
3月前
|
消息中间件 存储 数据库
深入学习RocketMQ的底层存储设计原理
文章深入探讨了RocketMQ的底层存储设计原理,分析了其如何通过将数据和索引映射到内存、异步刷新磁盘以及消息内容的混合存储来实现高性能的读写操作,从而保证了RocketMQ作为一款低延迟消息队列的读写性能。
|
3月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配:掌握消息路由的艺术!
【8月更文挑战第24天】在消息队列(MQ)体系中,交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器,它通过模式匹配实现消息的精准分发,适用于发布-订阅模式。不同于直接交换器和扇形交换器,`topic`交换器支持更复杂的路由策略,通过带有通配符(如 * 和 #)的模式字符串来定义队列与交换器间的绑定关系。
55 2
|
3月前
|
消息中间件
快来体验 消息队列RabbitMQ版入门训练营 打卡学习领好礼
快来体验 消息队列RabbitMQ版入门训练营 打卡学习领好礼
61 0
|
6月前
|
消息中间件 存储 数据安全/隐私保护
深入学习RabbitMQ五种模式(一)
深入学习RabbitMQ五种模式(一)
72 0
|
5月前
|
消息中间件 IDE 数据库
RocketMQ事务消息学习及刨坑过程
RocketMQ事务消息学习及刨坑过程
|
6月前
|
消息中间件 存储 负载均衡
消息队列学习之RabbitMQ
【4月更文挑战第3天】消息队列学习之RabbitMQ,一种基于erlang语言开发的流行的开源消息中间件。
54 0
|
6月前
|
消息中间件 存储 监控
写了10000字:全面学习RocketMQ中间件
以上是 V 哥在授课时整理的全部 RocketMQ 的内容,在学习时重点要理解其中的含义,正所谓知其然知其所以然,希望这篇文章可以帮助兄弟们搞清楚RocketMQ的来龙去脉,必竟这是一个非常常用的分布式应用的中间件,好了,今天的内容就分享到这,我靠!已经 00:36分,建议收藏起来,慢慢消化,创作不易,喜欢请点赞转发。
694 0
|
6月前
|
消息中间件 存储 缓存
消息队列学习之rocketmq
【4月更文挑战第1天】消息队列学习之rocketmq
45 0