RabbitMQ之基础入门

在 AMQP 中，Producer 将消息发送到 Exchange ，再由 Exchange 将消息路由到一个或多个 Queue 中（或者丢弃）。Exchange 根据 Routing Key 和 Binding Key 将消息路由到 Queue ，目前提供了 Direct、Topic、Fanout、Headers 四种类型。
RabbitMQ 概念
A. Queue
Queue: Queue（队列）是RabbitMQ的内部对象，用于存储消息,RabbitMQ 中的消息都只能存储在 Queue 中,生产消息并最终投递到Queue中, 消费者可以从Queue中获取消息并消费
在这里插入图片描述
B. Exchange
Exchange: 交换器，实际上这在 RabbitMQ 中是不会生产者将消息投递到队列中，而是，生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）在这里插入图片描述
C. Routing key
Routing key: 生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。
在这里插入图片描述
D. Binding

Binding: RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

E. Binding key

Binding key: 在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

交换机类型

  1. Direct Exchange
    Direct 类型的 Exchange 路由规则比较简单，它会把消息路由到那些 binding key 与 routing key 完全匹配的 Queue 中
    在这里插入图片描述
    我们以 routingKey=“key1” 发送消息到 Direct Exchange ，则消息会路由到 Queue1
    我们以 routingKey=“key2” 或者 routingKey=“key3” 发送消息到 Direct Exchange ，则消息会路由到 Queue2
    如果我们以其它 routingKey 发送消息，则消息不会路由到这两个 Queue 中。
    2. Topic Exchange
    Direct Exchange路由规则，是完全匹配 binding key 与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。 Topic Exchange 在匹配规则上进行了扩展，它与 Direct 类型的Exchange 相似，也是将消息路由到 binding key 与 routing key 相匹配的 Queue 中，但这里的匹配规则有些不同，它约定：

routing key 为一个句点号 “.” 分隔的字符串。我们将被句点号"."分隔开的每一段独立的字符串称为一个单词，例如 “pay.order.key”
binding key 与 routing key 一样也是句点号 “.” 分隔的字符串。
binding key 中可以存在两种特殊字符 * 与 #，用于做模糊匹配。其中 * 用于匹配一个单词，# 用于匹配多个单词（可以是零个）。
在这里插入图片描述
以上图为例

routingKey=“xx.key1.yy” 的消息会同时路由到 Q1 。
routingKey=“xx.key1.key2” 的消息会同时路由到 Q1 与 Q2 。
routingKey=“xx.yy.key2” 的消息会路由到 Q2 。
routingKey=“key3.xx.key2” 的消息会路由到Q2（只会投递给 Q2 一次，虽然这个 routingKey 与 Q2 的两个 bindingKey 都匹配）。
routingKey=“xx.yy.zz”、routingKey=“key1”、routingKey=“xx.key1.zz.key2” 的消息将会被丢弃，因为它们没有匹配任何 bindingKey 。
3. Fanout Exchange
Fanout Exchange 路由规则非常简单，它会把所有发送到该 Exchange 的消息路由到所有与它绑定的 Queue 中
在这里插入图片描述
生产者发送到 Exchange 的所有消息都会路由到图中的两个 Queue，并最终被消费者消费。
总结来说，指定 Exchange ，会路由到多个绑定的 Queue 中。
4. Headers Exchange

Headers Exchange 不依赖于 routing key 与 binding key 的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。

在绑定 Queue 与 Exchange 时指定一组 headers 键值对。
当消息发送到 Exchange 时，RabbitMQ 会取到该消息的 headers（也是一个键值对的形式），对比其中的键值对是否完全匹配 Queue 与 Exchange 绑定时指定的键值对；如果完全匹配则消息会路由到该 Queue ，否则不会路由到该 Queue 。
下面进行实战操作
A:Direct模式

confi配置类
@Configuration
public class DirectExchangeConfiguration {/*** 创建一个 Queue** @return Queue*/@Beanpublic Queue queue01() {// Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除return new Queue(Message01.QUEUE,true,false,false);}/*** 创建 Direct Exchange** @return DirectExchange*/@Beanpublic DirectExchange exchange01() {// name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它return new DirectExchange(Message01.EXCHANGE,true,false);}/*** 创建 Binding* Exchange：Message01.EXCHANGE* Routing key：Message01.ROUTING_KEY* Queue：Message01.QUEUE** @return Binding*/@Beanpublic Binding binding01() {return BindingBuilder.bind(queue01()).to(exchange01()).with(Message01.ROUTING_KEY);}
}
====================================》direct 类型的消息对象
@Data
public class Message01 implements Serializable {public static final String QUEUE = "QUEUE_01";public static final String EXCHANGE = "EXCHANGE_01";public static final String ROUTING_KEY = "ROUTING_KEY_01";private String id;
}
===============================>生产者
@Component
public class Producer01 {@Resourceprivate RabbitTemplate rabbitTemplate;public void syncSend(String id) {// 创建 Message01 消息Message01 message = new Message01();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Message01.EXCHANGE, Message01.ROUTING_KEY, message);}public void syncSendDefault(String id) {// 创建 Message01 消息Message01 message = new Message01();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Message01.QUEUE, message);}@Asyncpublic ListenableFuture<Void> asyncSend(String id) {try {// 发送消息this.syncSend(id);// 返回成功的 Futurereturn AsyncResult.forValue(null);} catch (Throwable ex) {// 返回异常的 Futurereturn AsyncResult.forExecutionException(ex);}}
==============================================>消费者
@Component
@RabbitListener(queues = Message01.QUEUE)
@Slf4j
public class Consumer01 {public void onMessage(Message01 message) {log.info("[Consumer01 onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}
}
======================================》direct的测试类@ResourceProducer01 producer01;@SneakyThrows@Testvoid syncSend() {String id = UUID.randomUUID().toString();producer01.syncSend(id);log.info("[test producer01 syncSend][id:{}] 发送成功", id);TimeUnit.SECONDS.sleep(2);}@Testvoid syncSendDefault() throws InterruptedException {String id = UUID.randomUUID().toString();producer01.syncSendDefault(id);log.info("[test producer01 syncSendDefault][id:{}] 发送成功", id);TimeUnit.SECONDS.sleep(2);}@Testvoid asyncSend() throws InterruptedException {String id = UUID.randomUUID().toString();producer01.asyncSend(id).addCallback(new ListenableFutureCallback<Void>() {@Overridepublic void onFailure(Throwable e) {log.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);}@Overridepublic void onSuccess(Void aVoid) {log.info("[testASyncSend][发送编号：[{}] 发送成功，发送成功]", id);}});log.info("[test producer01 asyncSend][id:{}] 发送成功", id);TimeUnit.SECONDS.sleep(2);}

B:Fanout 扇出 广播模式

==============》Fanout Exchange 示例的配置类
@Configuration
public class FanoutExchangeConfiguration {/*** 创建 Queue A** @return Queue*/@Beanpublic Queue queue03A() {// Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除return new Queue(Message03.QUEUE_A,true,false,false);}/*** 创建 Queue B** @return Queue*/@Beanpublic Queue queue03B() {return new Queue(Message03.QUEUE_B,true,false,false);}/*** 创建 Fanout Exchange** @return FanoutExchange*/@Beanpublic FanoutExchange demo03Exchange() {// name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它return new FanoutExchange(Message03.EXCHANGE,true,false);}/*** 创建 Binding A* Exchange：Message03.EXCHANGE* Queue：Message03.QUEUE_A** @return Binding*/@Beanpublic Binding demo03BindingA() {return BindingBuilder.bind(queue03A()).to(demo03Exchange());}/*** 创建 Binding B* Exchange：Message03.EXCHANGE* Queue：Message03.QUEUE_B** @return Binding*/@Beanpublic Binding demo03BindingB() {return BindingBuilder.bind(queue03B()).to(demo03Exchange());}
============================》Fanout Exchange 类型消息
@Data
public class Message03 implements Serializable {public static final String QUEUE_A = "QUEUE_03_A";public static final String QUEUE_B = "QUEUE_03_B";public static final String EXCHANGE = "EXCHANGE_03";private String id;
}
==============================》生产者
@Component
public class Producer03 {@Resourceprivate RabbitTemplate rabbitTemplate;/*** @param id         消息内容* @param routingKey key*/public void syncSend(String id) {Message03 message = new Message03();message.setId(id);rabbitTemplate.convertAndSend(Message03.EXCHANGE, null, message);}
==============================》消费者
@Component
@Slf4j
public class Consumer03 {@RabbitListener(queues = Message03.QUEUE_A)public void onMessage1(Message03 message) {log.info("[Consumer03 onMessage1][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}@RabbitListener(queues = Message03.QUEUE_B)public void onMessage2(Message03 message) {log.info("[Consumer03 onMessage2][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}
}
=====================================》Fanout Exchange 测试类@ResourceProducer03 producer03;@Testvoid syncSend() throws InterruptedException {String id = UUID.randomUUID().toString();producer03.syncSend(id);log.info("[test producer03 syncSend][id: {}] 发送成功", id);TimeUnit.SECONDS.sleep(2);}

C:Topic 主题模式

========================》TopicExchange 配置
@Configuration
public class TopicExchangeConfiguration {/*** 创建 Queue** @return Queue*/@Beanpublic Queue queue02() {// Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除return new Queue(Message02.QUEUE,true,false,false);}/*** 创建 Topic Exchange** @return TopicExchange*/@Beanpublic TopicExchange exchange02() {// name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它return new TopicExchange(Message02.EXCHANGE,true,false);}/*** 创建 Binding* Exchange：Message02.EXCHANGE* Routing key：Message02.ROUTING_KEY* Queue：Message02.QUEUE** @return Binding*/@Beanpublic Binding binding02() {return BindingBuilder.bind(queue02()).to(exchange02()).with(Message02.ROUTING_KEY);}
===========================================》topic 类型的实体类
@Data
public class Message02 implements Serializable {public static final String QUEUE = "QUEUE_02";public static final String EXCHANGE = "EXCHANGE_02";public static final String ROUTING_KEY = "#.key2.key3";private String id;
}
==============================================》生产者 topic 类型
@Component
public class Producer02 {@Resourceprivate RabbitTemplate rabbitTemplate;/*** @param id         消息内容* @param routingKey key*/public void syncSend(String id, String routingKey) {Message02 message = new Message02();message.setId(id);rabbitTemplate.convertAndSend(Message02.EXCHANGE, routingKey, message);}
}
===================================》 topic 消费者
@Component
@RabbitListener(queues = Message02.QUEUE)
@Slf4j
public class Consumer02 {@RabbitHandlerpublic void onMessage(Message02 message) {log.info("[Consumer02 onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}
}
=============================================》topic测试类@ResourceProducer02 producer02;@Testvoid syncSendSuccess1() throws InterruptedException {String id = UUID.randomUUID().toString();// 必须要满足  routingKey 规则才能够成功投递String routingKey = "aa.key2.key3";producer02.syncSend(id, routingKey);log.info("[test producer02 syncSendSuccess1][routingKey: {}][id: {}] 发送成功", routingKey, id);TimeUnit.SECONDS.sleep(2);}@Testvoid syncSendSuccess2() throws InterruptedException {String id = UUID.randomUUID().toString();String routingKey = "aa.bb.key2.key3";producer02.syncSend(id, routingKey);log.info("[test producer02 syncSendSuccess2][routingKey: {}][id: {}] 发送成功", routingKey, id);TimeUnit.SECONDS.sleep(2);}@Testvoid syncSendFail() throws InterruptedException {String id = UUID.randomUUID().toString();String routingKey = "aa.key2.zz";producer02.syncSend(id, routingKey);log.info("[test producer02 syncSendFail][routingKey:{}][id:{}] 发送成功", routingKey, id);TimeUnit.SECONDS.sleep(2);}

以上的是入门的基础的常见部分代码 若需完整代码 可识别二维码后 给您发代码。
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/7716.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

github提交不了的问题

github提交不了的问题

开了VPN提交的时候提示这个报错 是需要这两个端口号一致&#xff0c;就能提交了
阅读更多...
线程池

线程池

文章目录 普通线程池自定义线程池 普通线程池 //创建线程池&#xff0c;不指定上限 ExecutorService pool1 Executors.newCachedThreadPool(); //创建线程池&#xff0c;指定上限 ExecutorService pool1 Executors.newFixedThreadPool(5); --------------------------------…
阅读更多...
Material Studio 计算分子静电力、电荷密度以及差分电荷密度

Material Studio 计算分子静电力、电荷密度以及差分电荷密度

1.先打开Material Studio导入要计算的分子cif文件或者mol文件&#xff0c;直接Flie-Import 2.高斯几何优化一下结构&#xff0c;参数按照我的设置就行&#xff0c;一般通用&#xff0c;后面出问题再调整 3.点完Run后会跳出很多计算过程&#xff0c;不用管&#xff0c;等他计算完…
阅读更多...
【FL常用插件#1】Ozone11臭氧的安装和使用

【FL常用插件#1】Ozone11臭氧的安装和使用

本文内容收集自互联网&#xff0c;仅供个人学习参考使用&#xff0c;不允许用于商业用途&#xff0c;造成的侵权行为与本文作者无关 安装 VST2、VST3、AAX和NKS是音频技术界常见的几种插件格式&#xff0c;它们在功能和兼容性上有所不同&#xff1a; VST2 (Virtual Studio Tec…
阅读更多...
python环境下labelImg图片标注工具的使用

python环境下labelImg图片标注工具的使用

labelimg GitHub地址 python环境下labelImg图片标注工具的使用 1. 写在开头2. 如何使用2.1安装2.2 启动2.2.1 先启动后设置标注的目录2.2.2 指定标注的目录和预设置的标签 2.3 设置自动保存和显示类别。2.4 保存文件类型2.5 [快捷键](https://github.com/HumanSignal/labelImg…
阅读更多...
FastDFS-单机扩容

FastDFS-单机扩容

描述 周一上班收到用户反馈系统异常&#xff0c;紧急排查日志发现报错&#xff1a;FdfsServerException:错误:28&#xff0c;错误信息:没有足够的存储空间。 解决 根据异常信息判断是文件服务器可用内存不够了&#xff0c;首先登录文件服务器&#xff0c;使用df -h命令查看一…
阅读更多...
5.合并两个有序数组

5.合并两个有序数组

文章目录 题目简介题目解答解法一 &#xff1a;合并后排序解法二&#xff1a;双指针排序 题目链接 大家好&#xff0c;我是晓星航。今天为大家带来的是 合并两个有序数组 相关的讲解&#xff01;&#x1f600; 题目简介 题目解答 解法一 &#xff1a;合并后排序 假设我们要合…
阅读更多...
NXP i.MX8系列平台开发讲解 - 3.11 Linux PCIe设备调试(WIFI模块)

NXP i.MX8系列平台开发讲解 - 3.11 Linux PCIe设备调试(WIFI模块)

专栏文章目录传送门&#xff1a;返回专栏目录 文章目录 目录 1. WIFI 模块简单介绍 2. 设备驱动原理介绍 3. PCIE WIFI驱动实例分析 3.1 查看设备树 3.2 wifi 设备驱动代码分析 3.3 内核配置选项 4. WIFI驱动调试相关 根据前面对PCIe的讲解&#xff0c;对PCIe的整体都有…
阅读更多...
练习题（2024/5/7）

练习题（2024/5/7）

1验证二叉搜索树 给你一个二叉树的根节点 root &#xff0c;判断其是否是一个有效的二叉搜索树。 有效 二叉搜索树定义如下&#xff1a; 节点的左 子树 只包含 小于 当前节点的数。节点的右子树只包含 大于 当前节点的数。所有左子树和右子树自身必须也是二叉搜索树。 示例 …
阅读更多...
c++ cpp 在类中执行线程 进行恒定计算

c++ cpp 在类中执行线程 进行恒定计算

在编程中&#xff0c;顺序执行是常见的模式&#xff0c;但是对cpu的利用率不是很高&#xff0c;采用线程池&#xff0c;又太麻烦了&#xff0c;原因是还得不断地把任务拆分&#xff0c;扫描返回值。 如果 初始化n个类的时候&#xff0c;传递数据自身即可异步计算&#xff0c;那…
阅读更多...
JavaWEB 框架安全：Spring 漏洞序列.（CVE-2022-22965）

JavaWEB 框架安全：Spring 漏洞序列.（CVE-2022-22965）

什么叫 Spring 框架. Spring 框架是一个用于构建企业级应用程序的开源框架。它提供了一种全面的编程和配置模型&#xff0c;可以简化应用程序的开发过程。Spring 框架的核心特性包括依赖注入&#xff08;Dependency Injection&#xff09;、面向切面编程&#xff08;Aspect-Or…
阅读更多...
AIGC绘画辅助网站

AIGC绘画辅助网站

Midjourney风格样式 Midjourney Style Classifier | Andrei Kovalevs Midlibrary
阅读更多...
124.反转链表（力扣）

124.反转链表（力扣）

题目描述 代码解决&#xff08;思路1&#xff1a;双指针&#xff09; class Solution { public:ListNode* reverseList(ListNode* head) {ListNode*temp;//保存cur下一个节点ListNode*curhead;ListNode*preNULL;while(cur){tempcur->next;// 保存一下 cur的下一个节点&#…
阅读更多...
接口自动化测试的最佳实践和常见错误！

接口自动化测试的最佳实践和常见错误！

引言&#xff1a; 随着软件开发的不断演进&#xff0c;接口自动化测试在软件质量保障的过程中扮演着越来越重要的角色。良好的接口自动化测试可以提升测试效率、减少人力成本&#xff0c;但同时也存在着一些常见错误。本文将从零到一详细介绍接口自动化测试的最佳实践和常见错…
阅读更多...
突破传统 重新定义：3D医学影像PACS系统源码（包含RIS放射信息）实现三维重建与还原

突破传统 重新定义：3D医学影像PACS系统源码（包含RIS放射信息）实现三维重建与还原

突破传统&#xff0c;重新定义PACS/RIS服务,洞察用户需求&#xff0c;关注应用场景&#xff0c;新一代PACS/RIS系统&#xff0c;系统顶层设计采用集中分布式架构&#xff0c;满足医院影像全流程业务运行,同时各模块均可独立部署&#xff0c;满足医院未来影像信息化扩展新需求、…
阅读更多...
Ecovadis认证是什么？

Ecovadis认证是什么？

Ecovadis认证是一种企业社会责任&#xff08;CSR&#xff09;评估和评级的认证&#xff0c;旨在衡量企业在环境、劳工和人权、道德以及可持续采购四个方面的可持续发展表现。该认证已成为全球范围内许多公司和组织的评估标准之一&#xff0c;有助于提高企业的会声誉和可持续发展…
阅读更多...
Linux专栏10：Linux权限详解（上）

Linux专栏10：Linux权限详解（上）

博客主页&#xff1a;Duck Bro 博客主页系列专栏&#xff1a;Linux专栏关注博主&#xff0c;后期持续更新系列文章如果有错误感谢请大家批评指出&#xff0c;及时修改感谢大家点赞&#x1f44d;收藏⭐评论✍ Linux权限详解&#xff08;上&#xff09; 编号&#xff1a;10 文章…
阅读更多...
商务分析方法与工具（四）：Python的趣味快捷-简单函数你真的会用吗？

商务分析方法与工具（四）：Python的趣味快捷-简单函数你真的会用吗？

Tips&#xff1a;"分享是快乐的源泉&#x1f4a7;&#xff0c;在我的博客里&#xff0c;不仅有知识的海洋&#x1f30a;&#xff0c;还有满满的正能量加持&#x1f4aa;&#xff0c;快来和我一起分享这份快乐吧&#x1f60a;&#xff01; 喜欢我的博客的话&#xff0c;记得…
阅读更多...
SolidWorks进行热力学有限元分析一、模型建立

SolidWorks进行热力学有限元分析一、模型建立

1.话不多说按照我的操作来 2.这一步鼠标移到中心点直接拉就行 3.这里选单位&#xff0c;继续按照操作来 4.选中这个边&#xff0c;直接拉&#xff0c;输入尺寸后确定&#xff0c;其他边同理 5.鼠标右键设置厚度 6.右键零件&#xff0c;然后编辑材料&#xff0c;给他赋予你需要的…
阅读更多...
游戏辅助 -- 实战找人物对象基址

游戏辅助 -- 实战找人物对象基址

本节课在线学习视频&#xff1a; https://pan.quark.cn/s/3e83f4568031 一、打开CE工具&#xff0c;加载游戏进程 二、搜索人物血量144&#xff0c;选择首次扫描 三、进入游戏&#xff0c;让人物血量发生变化&#xff0c;搜索减少的数值 四、发现绿色的数值&#xff0c;一般绿…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023