RabbitMQ之基础入门

embedded/2024/10/18 7:56:27/

在 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.ppmy.cn/embedded/32981.html

相关文章

Typora+Docsify快速入门

Typora+Docsify快速入门

Typora是什么&#xff1f; ​ Typora中文版是一款好用极简的跨平台Markdown编辑器&#xff0c;软件使用这款软件能够帮助用户轻松将文本转换到HTML&#xff0c;软件从底层向上设计&#xff0c;软件支持markdown的标准语法&#xff0c;同时这款软件还支持动态预览功能&#xff…
阅读更多...
vCLS 和 FSVM代理虚拟机的默认用户名和密码。

vCLS 和 FSVM代理虚拟机的默认用户名和密码。

vCLS 代理虚拟机由 vSphere 集群服务自动生成并用于维护集群服务的运行状况。默认情况下&#xff0c;当集群内主机大于等于三台时&#xff0c;vCLS 虚拟机为三个&#xff0c;若集群内主机数小于三台时&#xff0c;vCLS 虚拟机等于主机的数量。这些代理虚拟机由 vSphere ESX Age…
阅读更多...
nginx配置tcp长连接实现集群

nginx配置tcp长连接实现集群

注意&#xff1a;实际工程应该会用docker部署。 安装nginx sudo apt install libpcre3-dev zlib1g openssl -y wget https://nginx.org/download/nginx-1.26.0.tar.gz #安装到/home/gyl/workspace/mprpc/vendor/nginx-1.26.0下 tar xfzv nginx-1.26.0.tar.gz && cd n…
阅读更多...
Python进阶之-ast使用详解

Python进阶之-ast使用详解

✨前言&#xff1a; &#x1f31f;什么是ast? ast模块在Python中用于将源码转换成抽象语法树&#xff08;Abstract Syntax Trees&#xff0c;AST&#xff09;。通过AST&#xff0c;我们可以读取、修改、分析Python代码。本质上&#xff0c;它将源码转化为树形结构&#xff0…
阅读更多...
【玩转Google云】GCP Kubernetes Engine (GKE) 深入解析

【玩转Google云】GCP Kubernetes Engine (GKE) 深入解析

Google Kubernetes Engine (GKE) 作为 Google Cloud Platform (GCP) 提供的托管式 Kubernetes 服务，为开发者和运维人员提供了一条通往云原生应用的便捷之路。本文将深入剖析 GKE 的核心优势和主要功能，带您领略其如何简化 Kubernetes 管理、提升应用可靠性与可扩展性，并保障…
阅读更多...
VMware虚拟机中ubuntu使用记录（6）—— 如何标定单目相机的内参（张正友标定法）

VMware虚拟机中ubuntu使用记录（6）—— 如何标定单目相机的内参（张正友标定法）

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、张正友相机标定法1. 工具的准备2. 标定的步骤(1) 启动相机(2) 启动标定程序(3) 标定过程的操作(5)可能的报错 3. 标定文件内容解析 前言 张正友相机标定法…
阅读更多...
JVM笔记1--Java内存区域

JVM笔记1--Java内存区域

1、运行时数据区域 从上图可以看出来&#xff0c;Java虚拟机运行时数据区域整体上可以分成5大块&#xff1a; 1.1、程序计数器 程序计数器是一块较小的内存空间。它可以看做当前线程所执行的字节码的行号指示器。在Java虚拟机的概念模型里&#xff0c;字节码解释器工作时就是…
阅读更多...
如何打包Apk适配32和64位

如何打包Apk适配32和64位

一个表格了解lib下的文件夹 .so文件描述armeabi-v7a第七代及以上的ARM处理器&#xff0c;2011年以后生产的大部分Android设备都使用。arm64-v8a第8代、64位ARM处理器&#xff0c;很少设备&#xff0c;三星GalaxyS6是其中之一。armeabi第5代、第6代的ARM处理器&#xff0c;早期…
阅读更多...
最新文章

Copyright @ 2022~2023