RabbitMQ学习笔记

RabbitMQ

中间件

在操作系统与应用软件之间搭建的桥梁。负责两者之间的通讯。(跨平台)

背景：微服务架构，若微服务之间需要通讯，此时就需要中间件为中介，并且保证持久性、高可用（防止服务挂掉）容错性，稳定性

分布式架构优劣：

存在问题

学习成本高，技术栈过多
运维成本和服务器成本增高、人员成本
项目的负载度也会上升
面临的错误和容错性也会成倍增加
占用的服务器端口和通讯的选择的成本高
安全性的考虑和因素逼迫可能选择RMI/MQ相关的服务器端通讯。

好处

服务系统的独立，占用的服务器资源减少和占用的硬件成本减少，
确切的说是：可以合理的分配服务资源，不造成服务器资源的浪费
系统的独立维护和部署，耦合度降低，可插拔性。
系统的架构和技术栈的选择可以变的灵活（而不是单纯的选择java）
弹性的部署，不会造成平台因部署造成的瘫痪和停服的状态。

选择一个优秀的消息中间件：

是否可以通讯
是否高可用 (集群)
是否持久化
是否跨平台（跨语言、操作系统）
容错性
可靠性

所以，消息队列是啥：

在跨服务、系统之间传递消息。并提供消息排队、消息传递机制

应用场景：

将任务异步处理，减少系统响应时间 – 高并发异步削峰
跨系统传递消息
分布式事务…

消息队列主要任务：接收、分发、存储消息。

保证高可靠：

消息传输机制：订阅、轮询、公平、可重传、拉取
支持集群
消息存储是否可靠：持久化

持久化：

数据存入磁盘，可永久保存

ActiveMQ	RabbitMQ	Kafka	RocketMQ
文件存储	支持	支持	支持	支持
数据库	支持	/	/	/

消息分发策略机制

发布订阅、轮询分发、公平分发、重发、消息拉取（类似于Git pull、push）

	ActiveMQ	RabbitMQ	Kafka	RocketMQ
发布订阅	✅	✅	✅	✅
轮询分发	✅	✅	✅	❌
公平分发	❌	✅	✅	❌
重发	✅	✅	❌	✅
消息拉取	❌	✅	✅	✅

RabbitMQ 消息的分发策略_duyinboo的博客-CSDN博客

发布订阅：生产者将消息发到消息队列，消费者订阅了后，就会收到消息队列的消息，<推>

轮询分发：

生产者将消息投递到消息队列，消息队列会按照一定的机制将消息推给消费者，（公平分发）。
自动应答

不会因为服务器性能的不同而产生的数据倾斜。每个服务收到的消息平均分-- 公平

公平分发：

按照服务器的性能分配，快的消费多，慢的消费少（能者多劳），会产生数据倾斜现象。
需要写程序手动应答

轮询分发与公平分发：共同点
- 若有消费者消费了消息，则该消息不会被其他消费者消费（不会重复消费）

重发：消息队列发送消息后，存在应答机制：若没有收到消费者的成功反馈，则重新发送给其他消费者(服务器) 消费，防止消息堆积。保证消息可靠性。

消息拉取： <拉>

安装

(71条消息) windows环境下安装RabbitMQ（超详细）_luckySnow-julyo的博客-CSDN博客

Windows 10 下 RabbitMQ的安装 - 知乎 (zhihu.com)

消息模型

RabbitMQ6中模型：

基本消息模型：一个生产者生产消息，放入消息队列，一个消费者从消息队列中接收消息
工作模型：一个生产者，将消息放入队列，多个消费者接收消息。一个消息只能被一个消费者消费。
1. 点对点模型。场景：订单系统、日志处理。、
订阅模型：一个生产者，多个生产者。生产者将消息发给交换机，由交换机决定消息的发送模式；每个队列需要绑定交换机。一个队列关联一个消费者。
1. 广播：交换机将消息发送给绑定了交换机的所有队列。
2. 路由：交换机通过不同的路由key 将消息发送到不同的队列。
  1. eg：队列1 路由key 为 error，只接收交换机发送的error的消息；队列2 路由key 为 warning，只接收交换机发送的warning的消息
3. Topic 主题：在路由的基础上，队列接收符合通配符的消息。路由通过模糊匹配，将消息发送到匹配的队列。

RabbitMQ优缺点

优点：

应用解耦；降低应用间的耦合度，不需要再修改代码
异步提速，减少系统响应时间
削峰：将高峰时期的消息暂存至MQ（削峰），系统定期从MQ中消费自己能承受的消息量（填谷）。
1. 提高系统可用性

劣势：

系统可用性降低；（多加了一个模块）
1. 若 MQ 挂掉，则消息丢失，如何保证MQ高可用
系统复杂性提高
1. MQ异步调用消息，怎么保证消息
  1. 是否被重复消费？
  2. 消息丢失问题
  3. 消息传递顺序性？
一致性问题
1. 若某个消费者消费消息失败；则应如何保证 数据一致性

应用场景

消费者处理消息后不需要返回结果；
接收短时间的数据不一致；

数据一致性要求很高、第三方支付、银行转账等不适用MQ

Rabbitmq架构

在这里插入图片描述

Broker：接收、分发消息
Virtual Host：当多个用户使用同一个MQ时，会分出多个Virtual Host ；每个 VH 包含各自的交换机和队列；
Exchange：交换机，通过匹配路由（routing key）不同类型的分发规则将消息传给不同的队列。
Queue：接收来自交换机的消息。传给消费者消费。
Binding：Exchange与 Queue的连接。包括 routing key，Binding 被保存在交换机的 查询表中
1. Exchange 的查询表，其中记录了 Exchange 的名称、类型、是否持久化等属性，以及与之绑定的队列和 RoutingKey 等信息。具体操作步骤如下：
  1. 进入 RabbitMQ 的管理界面；
  2. 在左侧导航栏中选择“Exchanges”选项；
  3. 在查询表中查看 Exchange 的相关信息。
Connection：客户端（生产、消费者）与 Broker 建立TCP连接；
channel：轻量级的Connection，避免频繁建立TCP连接，

Web管理页：

RabbitMQ 管理页面该如何使用 - 掘金

常见端口

(101条消息) 消息中间件RabbitMQ需要知道的6个端口的作用_rabbitmq 4369_像夏天一样热的博客-CSDN博客

管理界面的登陆页

http://127.0.0.1:15672 
1

端口	作用
15672	管理界面ui使用的端口
15671	管理监听端口
5672，5671	AMQP 0-9-1 without and with TLSclient端通信口
4369	（epmd)epmd代表 Erlang端口映射守护进程，erlang发现口
25672	( Erlang distribution） server间内部通信口

常用命令：

rabbitmq 常用命令

启动监控管理器：rabbitmq-plugins enable rabbitmq_management
关闭监控管理器：rabbitmq-plugins disable rabbitmq_management
启动rabbitmq：rabbitmq-service start
关闭rabbitmq：rabbitmq-service stop
查看所有的队列：rabbitmqctl list_queues
清除所有的队列：rabbitmqctl reset
关闭应用：rabbitmqctl stop_app
启动应用：rabbitmqctl start_app
用户和权限设置（后面用处）

添加用户：rabbitmqctl add_user username password
分配角色：rabbitmqctl set_user_tags username administrator
新增虚拟主机：rabbitmqctl add_vhost  vhost_name
将新虚拟主机授权给新用户：rabbitmqctl set_permissions -p vhost_name username '.*' '.*' '.*'
角色说明

none  最小权限角色
management 管理员角色
policymaker   决策者
monitoring  监控
administrator  超级管理员 

-----------------------------------------
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

1. 关闭与启动
① 到指定目录：cd/etc/init.d
② 停止：rabbitmq-server stop
③ 启动：rabbitmq-server start
④ 查看是否停止/启动成功：ps -ef |grep rabbitmq
2.开启RabbitMQ Managerment管理界面
① 到指定目录：cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.1.5/plugins
② 开启管理界面：./rabbitmq-plugins enable rabbitmq_management
1
2
3
4
5
6
7
8

Linux

启动 RabbitMQ 服务：

sudo systemctl start rabbitmq-server
1

停止 RabbitMQ 服务：

sudo systemctl stop rabbitmq-server
1

重启 RabbitMQ 服务：

sudo systemctl restart rabbitmq-server
1

查看 RabbitMQ 服务状态：

sudo systemctl status rabbitmq-server
1

启用 RabbitMQ 管理界面：

sudo rabbitmq-plugins enable rabbitmq_management
1

查看 RabbitMQ 管理界面地址：

sudo rabbitmqctl status | grep "listener.*{tcp.*http"
1

查看 RabbitMQ 队列列表：

sudo rabbitmqctl list_queues
1

删除 RabbitMQ 队列：

sudo rabbitmqctl delete_queue [QUEUE_NAME]
1

查看 RabbitMQ 交换机列表：

sudo rabbitmqctl list_exchanges
1

删除 RabbitMQ 交换机：

sudo rabbitmqctl delete_exchange [EXCHANGE_NAME]
1

windows

以下是 RabbitMQ 在 Windows 环境下的常用命令：

注意，在 Windows 环境下，RabbitMQ 相关命令需要在 RabbitMQ 安装目录下的 sbin 文件夹下执行。

启动 RabbitMQ 服务：

net start RabbitMQ
1

停止 RabbitMQ 服务：

net stop RabbitMQ
1

重启 RabbitMQ 服务：

net stop RabbitMQ
net start RabbitMQ
1
2

查看 RabbitMQ 服务状态：

sc query RabbitMQ
1

启用 RabbitMQ 管理界面：

rabbitmq-plugins enable rabbitmq_management
1

查看 RabbitMQ 管理界面地址：

rabbitmqctl status | findstr "http:"
1

查看 RabbitMQ 队列列表：

rabbitmqctl.bat list_queues
1

删除 RabbitMQ 队列：

rabbitmqctl.bat delete_queue [QUEUE_NAME]
1

查看 RabbitMQ 交换机列表：

rabbitmqctl.bat list_exchanges
1

删除 RabbitMQ 交换机：

rabbitmqctl.bat delete_exchange [EXCHANGE_NAME]
1

可视化界面管理：

启动

rabbitmq-plugins enable rabbitmq_management 
1

关闭

rabbitmq-plugins disable rabbitmq_management
1

MQ消息投递、ACK机制总结

RabbitMQ 消息投递以及ACK机制 - Mr*宇晨 - 博客园

RabbitMQ的消息模型

简单模式

一个生产者，一个消费者，默认交换机，一个队列。
在这里插入图片描述

消费者监听来自消息队列的消息。若队列中有消息，则消费；随后队列将该消息删除；
- 缺点：若消费者拿到消息后发生故障没有正常消费消息，会造成消息丢失。

工作模式

消息只能被消费一次。 可能会重复放入队列。

消费者c1，c2 轮询或非公平消费队列中的消息。（不会重复消费）

高并发下可能会重复放入队列：
解决方法见幂等性-重复消费

 在rabbitmq的工作队列模式下，在高并发情况下，某一条消息有可能会被重复放入队列。
  
    当且仅当
    	（消费者没及时确认消息，此时该消费者挂掉，会导致消息被重新放入队列，以至于重新消费该消息。）
 
 	要避免这种情况，可以考虑使用“任务分配模式”，让每个消费者通过channel.basicqos(1)设定取值比例，使得每个消费者只能顺序地获取一个消息来处理，这样就可以避免竞争和重复处理
1
2
3
4
5
6

在这里插入图片描述

RabbitMQ之Work Queues模式_

消息应答（消费者确认）

自动应答

手动确认

默认是轮询消费消息。

消息应答：默认情况下，手动消息确认是打开的。在前面的例子中，我们通过auto_ack=True标志显式地关闭了它们。

自动应答（消息发给消费者后自动删除消息。）

带来的问题：若消费者此时挂掉会造成消息丢失
解决方法：手动确认消息。
1. 让消费者消费消息完成后向队列发送消息确认，随后队列删除消息。
2. 若消费者挂掉，没有发送消息确认。则该消息重新被放入队列，可让其他消费者消费该消息。
3. RabbitMQ不会为未确认的消息设置超时时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久。仅当消费者挂掉才会将消息回退到队列中。
  
  rabbitmq.conf中的参数consumer_timeout自定义超时时间

手动确认：看实现资料 01.pdf 消费者

重写 defaultConsumer的 handleDelivery() 方法：

channel.basicConsume(将autoAck设为false。)

DefaultConsumer consumer = new DefaultConsumer(channel){ 
    @Override 
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
        String msg = new String(body); 
        int i = 5/0; 
        System.out.println("received:"+msg); 
        // 手动进行ACK 
        channel.basicAck(envelope.getDeliveryTag(),false);
    } 
};
	// 监听队列，第二个参数false，手动进行ACK 
	channel.basicConsume(QUEUE_NAME,false,consumer);
1
2
3
4
5
6
7
8
9
10
11
12

应答方法

deliveryTag: 发送消息的标签; multiple: 是否批量应答

multiple

true：批量应答；当前指针为8，则前面未应答的5,6,7都会被应答。
false：仅应答当前指针的消息。前面未应答的不会应答。

返回	方法	说明
void	`basicAck(long deliveryTag, boolean multiple)`	确认收到
void	`basicReject(long deliveryTag, boolean requeue)`	拒绝消息，并丢弃
void	`basicNack(long deliveryTag, boolean multiple, boolean requeue)`	拒绝收到

ACK确认与消息接收必须使用相同的channel信道

若消费者未确认消息，便将消费者关闭，则队列会回收消息导致重复消费。
消费者没关闭，会一直占用CPU内存。

Springboot 消费者确认：

RabbitMQ(4):消息确认机制详解 - 掘金

Springboot 的确认模式有三种，配置如下：

spring.rabbitmq.listener.simple.acknowledge-mode=manual
1

NONE ：不确认：
- 1、默认所有消息消费成功，会不断的向消费者推送消息
- 2、因为 rabbitmq 认为所有消息都被消费成功。所以队列中存在丢失消息风险。
AUTO：自动确认
- 1、根据消息处理逻辑是否抛出异常自动发送 ack（正常）和nack（异常）给服务端，如果消费者本身逻辑没有处理好这条数据就存在丢失消息的风险。
- 2、使用自动确认模式时，需要考虑的另一件事情就是消费者过载。
MANUAL：手动确认
- 1、手动确认在业务失败后进行一些操作，消费者调用 ack、nack、reject 几种方法进行确认，如果消息未被 ACK 则发送到下一个消费者或重回队列。
- 2、ack 用于肯定确认；nack 用于否定确认；reject 用于否定确认（一次只能拒绝单条消息）

队列持久化

消息持久化

队列持久化：不会将队列中的消息持久化！！！

channel.queueDeclare(); 将durable 设为 true。

若原先队列已设为 false，则需要删除原队列，重新queueDeclare
消息持久化：

channel.basicPublish() 将 props 设为 MessageProperties.PERSISTENT_TEXT_PLAIN

当然，也不是完全持久化，队列不一定就立刻刷入磁盘。
非公平分发：

存在多个消费者的效率不同，若采用轮询策略易导致消息堆积，消费者有的很清闲，有的很忙。此时需要进行公平分发（能者多劳。）
1. channel.basicQos(1) 消费者每次只能处理不能超过一个消息。
- 若消费者还没将上次的消息进行ACK，则不会将新的消息派给该消费者。
- 直到消费者ACK后，进行继续分发消息。
预期值：允许消费者未确认消息的最大数量。

channel.basicQos(6);

发布确认（生产者确认）

producer -> exchange -> queue -> consumer

broker

此部分是当消息被 queue 收到后，broker 向生产者确认消息（ 生产者丢失数据 ）

发布确认高级是确认 producer 到 broker (交换机确认)、exchange 到 queue 的消息（queue确认）。

发布确认：生产者通过channel发布的消息，当消息被对应的队列收到后，broker server会给生产者发送ACK确认包（消息唯一ID），确保消息已经正确到达目标队列。

若MQ队列持久化、消息持久化。
- 生产者通过channel将消息发给队列，当消息和MQ刷入磁盘后，broker 会告知生产者消息确认。
开启确认：

channel.confirmSelect();
1

代码详见 idea：rabbitmq-1-producter 项目下： work_producter_confirm.java

单个确认发布：当上一个消息 broker 返回确认后，消费者再发送下一个消息。
```
 boolean wait = channel.waitForConfirms();
1
```
缺点：逐个发布确认，可保证消息是否被 mq成功收到。但效率太慢
批量发布确认：每发送指定数量后同一确认。

if (i%confirmCount==0&&i>0){
    boolean wait = channel.waitForConfirms();
    if (wait){
        System.out.println("消息发送成功"+i);
    }
}
1
2
3
4
5
6

缺点：若出现问题，无法精确到是哪个消息没有被mq收到。

异步确认：利用回调函数 确定哪些消息成功接收、未成功接收。

在这里插入图片描述


  //支持高并发，且用跳表提高查询效率。
ConcurrentSkipListMap<Long, String> skipListMap = new ConcurrentSkipListMap<>();
    // 异步消息正常的确认
    ConfirmCallback ackCallback = (long deliveryTag, boolean multiple)->{
        // todo 批量删除  包括当前消息之前的所有消息 都是已确认的，需要删除
        // 剩下的就是 未正常确认的消息。
       	 	... 略
        // 不是批量 则单独删除 
       		... 略
    };
    // 异步消息异常的确认
    ConfirmCallback nackCallback = (long deliveryTag, boolean multiple)->{... };
    // 消息监听器，用于监听消息 接收是否成功
    channel.addConfirmListener(ackCallback,nackCallback);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// 发布消息时：
            skipListMap.put(channel.getNextPublishSeqNo(),message);
// 去掉 消息持久化 ackCallback就能打印出多个消息
            channel.basicPublish("",queueName, null,message.getBytes());
1
2
3
4

ConcurrentSkipListMap

J.U.C 之 ConcurrentSkipListMap - 掘金

(102条消息) ConcurrentSkipListMap 常用的方法_

headMap() 方法：

对返回的部分视图 map 的操作会映射到原Map上。

也就是说，这里获取headMap并clear，是清除消息队列Map中在该delivery Tag之前的所有消息，也就是批量确认

订阅发布

在 simple和work模式中消息只能被消费一次；

在这里插入图片描述

若想将一个消息发给多个消费者：由交换机 routingKey绑定多个队列，

交换机类型：

默认交换机：（AMQP default）
```
channel.basicPublish("","queueName",null,message.getBytes());
1
```
交换机名称默认为空字符串，将消息传给队列名称为 queueName的队列。

默认交换机 默认绑定到每个队列，路由键等于队列名称。

无法显式绑定到默认交换机或从默认交换机取消绑定。也无法删除。

临时队列：
- 即没有持久化的队列，断开与消费者的连接后，队列便会自动删除。
- 创建临时队列：
  - ```
  String queueName = channel.queueDeclare().getQueue();
  1
```
  特有的 Exclusive：used by only one connection and the queue will be deleted when that connection closes只被一个连接使用，当那个连接关闭时队列将被删除

bind：交换机与队列绑定

/**
 *  @param queue the name of the queue
 *  @param exchange the name of the exchange
 *  @param routingKey the routing key to use for the binding
 */
channel.queueBind(queue1,exchangeName,routingKey);
1
2
3
4
5
6

fanout 广播

在 fanout 模式下只要与声明的交换机绑定，不论多个队列routingKey 是否一样，都会将消息传给绑定了交换机的队列

消费者收到消息不会受到 RoutingKey的影响 , 只需绑定就可以收到通知

fanout类型的交换机试忽略routingkey的，只要绑定了队列都会收到消息
direct 不同的 routingKey 绑定不同的队列。
topics

星号 "*"：代替一个词
井号 "#"：代替零个或多个词

如果只有一个 # 那么将会接收通道的所有数据（fanout）

如果没有 #/* 出现 , 默认采用 direct

例子：

RoutingKey 通配值说明

com.sans.color *.sans.* 匹配3个单词中的中间单词 sans

com.sans.color.red #.red 匹配最后为 red

com.sans.color.blue con.# 匹配开头为 com

通配符规则：

在这里插入图片描述

死信队列：

RabbitMQ - Springboot 整合死信队列(Dead-Letter-Exchange ) - 知乎

应用场景

当消息消费发生异常时，将消息投入死信队列中。保证订单业务的消息数据不丢失

eg：用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

在 RabbitMQ 中使用 Confirm 模式时，异步未确认消息的处理方式取决于你的代码实现。一般来说，有两种处理方式：

使用publisher confirms机制中的 delivery-tag 概念，记录每一条消息发布的唯一编号（delivery tag），并保存到一个集合中。当消息被 RabbitMQ 确认（acknowledged）时，将 delivery tag 移除。在超时时间内（一般为几秒钟）如果 delivery tag 没有被移除，那么就认为这条消息发布失败了，再进行相应的处理。
定时检查未确认的消息集合，尝试重新发布这些消息，并更新 delivery tag。如果消息通过这种方式重新投递了多次仍然未能得到确认，那么就可以将它们转移到一个死信队列中，并在需要时进行相应的处理。

即：无法被消费的消息。某种原因导致消息无法被消费，若也没做后续处理，就会放入死信队列

导致的原因：

消息过期 TTL
队列满了（达到消息最大长度）
消息被拒绝

在这里插入图片描述

生产者发送10条消息，若此时C1 断链，且消息过期（ 10s ），消息就认为无法正常消费，便转到死信队列 dead-queue ,（若消息发太快即使未过期，但C1断链无法消费，10s内也会转到 dead-queue）。

普通队列与死信交换机绑定：

        Map<String, Object> params = new HashMap<>();
        // 指定死信交换机
        params.put("x-dead-letter-exchange",deadExchange);
        // 指定死信 routingKey
        params.put("x-dead-letter-routing-key","dead");

        // -- 设置 正常队列接收消息长度限制：
//        params.put("x-max-length",6);

//        由于正常的队列 需要与 死信交换机绑定：
        channel.queueDeclare(normalQueue,false,false,false,params);
1
2
3
4
5
6
7
8
9
10
11

设置 队列中所有消息的过期时长：

 // 生产者发送消息 设置消息过期时长：10s 
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

// 发布者发布消息
 channel.basicPublish(normalExchange,"normal",properties,message.getBytes());
1
2
3
4
5

设置正常队列长度限制

      // -- 设置 正常队列接收消息长度限制：
      params.put("x-max-length",6);
1
2

设置拒绝消息：

// 设置消费者 拒绝消息
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
// 取消手动接收
channel.basicConsume(dead_producter.normalQueue,
                             false,  deliverCallback,  cancelCallback);
1
2
3
4
5

延迟队列

一起来学SpringBoot | 第十三篇：RabbitMQ延迟队列-腾讯云开发者社区-腾讯云

使用场景：需要在某个事件发生之后或者之前的指定时间点完成某一项任务

1.订单在十分钟之内未支付则自动取消

2.新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。

3.用户注册成功后，如果三天内没有登陆则进行短信提醒。

4.用户发起退款，如果三天内没有得到处理则通知相关运营人员。

5.预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

设置TTL

消息设置：

对每条消息设置过期时长：

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

---------------------------------------------------
    @GetMapping("/hello")
    public void hello() {
        Message message = MessageBuilder.withBody("hello javaboy".getBytes())
                .setExpiration("10000")
                .build();
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_QUEUE_DEMO, message);
    }
作者：江南一点雨
链接：https://juejin.cn/post/7051469607806173221
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
1
2
3
4
5
6
7
8
9
10
11
12
13
14

队列设置：

在队列中统一设置消息时长： x-message-ttl 属性

@Bean("queueB")
public Queue queueB(){
    // 设置死信队列 和TTL
    Map<String, Object> arguments = new HashMap<>();
    arguments.put("x-dead-letter-exchange",exchangeY);
    arguments.put("x-dead-letter-routing-key","YD");
    arguments.put("x-message-ttl",30*1000);
    return  QueueBuilder.durable(queueB).withArguments(arguments).build();
}
1
2
3
4
5
6
7
8
9

对于第一种方式（每个消息设置自己的过期时间），当消息过期后并不会立马被删除，而是当消息要投递给消费者的时候才会去删除，因为每条消息的过期时间都不一样，想要知道哪条消息过期，必须要遍历队列中的所有消息才能实现，当消息比较多时这样就比较耗费性能，因此对于此方式，当消息要投递给消费者的时候才去删除。
对于第二种方式（在队列中设置统一的消息过期时间），当消息队列设置过期时间的时候，那么消息过期了就会被删除，因为消息进入 RabbitMQ 后是存在一个消息队列中，队列的头部是最早要过期的消息，所以 RabbitMQ 只需要一个定时任务，从头部开始扫描是否有过期消息，有的话就直接删除。

关于 RabbitMQ，应该没有比这更详细的教程了！ - 掘金

Delayed Message 插件

问题：queue只会计算头部消息时间，如果前面的消息ttl过长，后面就起不到准确的延时效果；易引起消息堆积。

解决方法：rabbitmq的插件：rabbitmq_delayed_message_exchange

原理：当消息发布时，先存入 Mnesia 分布式数据库管理系统中，当消息过期时，再开始发送消息到指定队列。

将消息ttl 放在交换机处：
在这里插入图片描述

直接将交换机设为延时；支持延时投递消息的机制。

(107条消息) @Component和@Configuration的差别_@configuration和@component区别_墨子白的博客-CSDN博客

延时队列和死信队列

延时队列（Delay Queue）：延时队列是一种特殊的队列，用于实现延时任务的功能。当消息发送到延时队列时，消息不会立即被消费者接收，而是会在指定的延时时间后再被消费者消费。延时队列可以用于实现各种延迟任务，例如延迟发送短信、延迟执行定时任务等。在RabbitMQ中，可以通过设置消息的过期时间，将消息发送到延时队列中。
死信队列（Dead Letter Queue）：死信队列是用于处理无法被消费者正确处理的消息的队列。当消息无法被消费者正确处理时，可以选择将其发送到死信队列中。这些无法处理的消息可以是过期消息、消费者拒绝接收消息、队列达到最大长度等情况下的消息。通过设置死信队列，可以对这些消息进行特殊处理，例如记录错误日志、发送警报通知等。在RabbitMQ中，可以通过设置消息的过期时间、消费者的拒绝等条件，将消息发送到死信队列中。

总结来说，延时队列用于实现延时任务的功能，而死信队列用于处理无法被正确处理的消息。它们在消息队列的应用中有不同的用途。

发布确认高级（生产者确认）

RabbitMQ(4):消息确认机制详解 - 掘金

producer -> broker -> exchange -> queue -> consumer 。

在编码时我们可以用两个选项用来控制消息投递的可靠性：

交换机确认：消息从 producer 到 RabbitMQ broker cluster 成功，则会返回一个 confirmCallback；
queue确认：消息从 exchange 到 queue 投递失败，则会返回一个 returnCallback

我们可以利用这两个 callback 接口来控制消息的一致性和处理一部分的异常情况。

confirmCallback

交换机确认

作用：判断 交换机是否接收到消息且对生产者做出反馈。

配置文件：
```
spring:
  rabbitmq:
    publisher-confirm-type: correlated
1
2
3
```
它有三个值：
- NONE：禁用发布确认模式，是默认值
- CORRELATED：发布消息成功到交换器后触发回调方法
- SIMPLE：值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;
  - 即：发一个消息，接收一个，确认后再继续发消息。（单个确认发布）

config 交换机、队列、 routingkey

@Configuration
@Slf4j
public class SeniorConfirmConfig {

    private final String queue = "SeniorConfirmQueue";
    private final String exchange = "SeniorConfirmExchange";
    private final String routingKey = "key1";

    @Bean
    public Queue SeniorQueue(){
        return QueueBuilder.durable(queue).build();
    }
    @Bean
    public DirectExchange SeniorExchange(){
        return ExchangeBuilder.directExchange(exchange).build();
    }

    @Bean
    public Binding bindingQE(@Qualifier("SeniorQueue") Queue  queue, @Qualifier("SeniorExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(routingKey);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

实现 RabbitTemplate 的接口： confirmCallback 以此交换机响应生产者消息。

创建rabbitTemplate 的实例Bean 后，init 实现的实例。

@PostConstruct：先执行完构造方法，再注入依赖，最后执行初始化操作

Spring框架@PostConstruct注解详解 - 掘金

@Component
@Slf4j
public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    // 创建rabbitTemplate 的实例Bean 后，init 实现的实例。
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
    }

    @Override
    /**
     *  交换机 回调方法
     *      correlationData 消息的信息、ID
     *      ack 交换机收到消息是否成功
     *      cause 交换机未收到消息的原因 （当 ack=false）
     */
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack){
            log.warn("交换机 已接收到 ID 为 ：{} 的消息",correlationData.getId());
        }else {
            log.error("交换机 未接收到 ID 为 ：{} 的消息，原因是：{}",correlationData.getId(),cause);

        }
    }
}
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

生产者、消费者

CorrelationData ：消息的额外消息（唯一标识）。 (id,Message)
作用：为 ConfirmCallback和ReturnCallback的回调函数设置消息的唯一标识。
1. 通过 CorrelationData的ID 找到对应消息；随后可判断消息是否被交换机接收

@GetMapping("sendSeniorConfirmMsg/{message}")
public void  sendSeniorConfirmMsg(@PathVariable String message){
    log.info("时间：{}；为队列SeniorConfirmQueue发送消息：{} ",new Date(),message);
    CorrelationData correlationData = new CorrelationData("1");
    correlationData.setReturnedMessage(new Message("returnedMessage".getBytes()));
    rabbitTemplate.convertAndSend("SeniorConfirmExchange","key1",message,correlationData);
}

--------------------
    //消费者：
    @RabbitListener(queues = "SeniorConfirmQueue")
    public void SeniorConfirmQueueMessage(Message message, Channel channel){
        System.out.println(message.getMessageProperties());
        System.out.println(channel);
        log.info("时间：{}，接收到来自{}的消息: {}",new Date(),
                 message.getMessageProperties().getConsumerQueue(),
                 new String(message.getBody()));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

returnedMessage

queue确认：

作用：当消息路由不可达（没有该 routingkey ）时，将消息告知生产者。

配置文件：

spring:
  rabbitmq:
    host: 127.0.0.1
    username: heima
    password: heima
    virtual-host: testheima
    port: 5672
    publisher-confirm-type: correlated
    publisher-returns: true
1
2
3
4
5
6
7
8
9

returnedMessage

当消息路由不可达（没有该 routingkey ）时，将消息告知生产者。

@Override
/**
 * Returned message callback.
 *   当消息路由 不可达（不存在该 routingkey ） 时，告知生产者。
 * @param message the returned message.
 * @param replyCode the reply code.
 * @param replyText the reply text.
 * @param exchange the exchange.
 * @param routingKey the routing key.
 */
    public void returnedMessage(Message message, int replyCode, 
                                String replyText, String exchange,
                                String routingKey) {
        log.error("消息 {} ，响应码 {} ， 回退原因 {} ， 交换机 {} ，routingKey {}",message,replyCode,replyText,exchange,routingKey);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

备份交换机

当某消息无法路由时，若通过 returnedMessage 告知生产者，通常为打印日志的方式；其实也可将消息存在另一个交换机上。由备份交换机处理

! 在这里插入图片描述

SeniorExchange 绑定备份交换机：backUpExchange

@Bean
public DirectExchange SeniorExchange(){
    return ExchangeBuilder.directExchange(seniorConfirmExchange)
            // 设置备份交换机
            .alternate(backUpExchange).build();
}
1
2
3
4
5
6

当 returnedMessage 与备份交换机同时开启，则优先进入备份的交换机。

幂等性

定义：用户对于同一操作发起一次或多次请求，返回的结果是一致的

RabbitMQ幂等性的主流解决方案 - 简书

RabbitMQ 消息幂等性&顺序性&消息积压&面试问题 - Mr*宇晨 - 博客园

RabbitMQ可能遇到的问题：

重复消费：

背景：

消费者正常接收到 MQ的消息，消费完后向MQ确认时（返回ACK）网络中断，导致MQ认为消费者未成功消费消息。MQ会把该消息发给其他消费者消费。从而导致重复消费。

解决方法：

唯一ID+ 指纹码判断是否被消费；使用数据库去重
1. 为消息生成唯一ID和指纹码作为数据库的主键；
  1. 消息消费之前，先在数据库中查询该消息主键是否存在：
    1. 不存在：未消费过。insert插入代表已消费过。
    2. 若存在，则已消费过；不再消费。
指纹码：通过规则或者时间戳加别的服务给到的唯一信息码，以确保消息唯一性。

缺点：高并发情况数据库读写压力大，可根据ID分库分表，采用路由算法分压分流。
使用Redis setnx 原子命令判断
1. 消息生成唯一ID 存入 Redis，若 setnx 成功则说明没消费过，
  1. 否则消息已被消费过。
使用 redis 的原子性去实现主要需要考虑两个点
- 第一：我们是否要进行数据落库，如果落库的话，关键解决的问题是数据库和缓存如何做到原子性？
- 第二：如果不进行落库，那么都存储到缓存中，如何设置定时同步的策略(同步到关系型数据库)？缓存又如何做到数据可靠性保障呢
  - 关于不落库，定时同步的策略，目前主流方案有两种，
    - 第一种为双缓存模式，异步写入到缓存中，也可以异步写到数据库，但是最终会有一个回调函数检查，这样能保障最终一致性，不能保证100%的实时性。
    - 第二种是定时同步，比如databus同步。

优先级队列

背景：重要客户的消息需要优先消费。

在队列中对消息进行排序，（设置优先级）再进行发给消费者。

官方的优先级范围： 0-255

设置队列中优先级范围：0-10

Map<String, Object> params = new HashMap();
params.put("x-max-priority", 10);
channel.queueDeclare("hello", true, false, false, params);
1
2
3

Boot版本：定义优先级队列 0-100

·  /**
     * 定义优先级队列
     */
    @Bean
    Queue queue() {
        Map<String, Object> args= new HashMap<>();
        args.put("x-max-priority", 100);
        return new Queue(QUEUE, false, false, false, args);
    }
1
2
3
4
5
6
7
8
9

消息代码设置优先级：

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();
1

Boot版本： 设置消息优先级

rabbitTemplate.convertAndSend("exchange","routringKey","message", 
                              (Message message2) ->{
                message2.getMessageProperties().setPriority(4);
                return message2;
}, correlationData1);
1
2
3
4
5

惰性队列

惰性队列会尽可能将消息存入磁盘，当消费者需要消费消息时，再将消息加载至内存。

解决的问题：消费者宕机导致消息堆积

两种模式：
default模式：MQ默认将消息放入内存，以便于将消息快速发给消费者。
- 即使消息持久化，内存也会存储部分消息。
lazy模式：将消息存入磁盘。
- 使用场景：
  1. 用于处理大量消息和长时间存储的场景：日志收集、大数据处理。
  2. 消费者宕机导致消息堆积
- 优点：只有在消费者需要消费时才会被加载到内存中，可以有效地降低内存的使用，提高系统的性能和稳定性。
- 缺点：会对消息的传输和消费带来一定的延迟

Map<String, Object> args = new HashMap<String, Object>(); 
args.put("x-queue-mode", "lazy"); 
channel.queueDeclare("myqueue", false, false, false, args);
1
2
3

镜像队列

Mirror Queue

RabbitMQ 镜像队列使用和原理详解 - 掘金

可在多个节点之间复制队列中的消息。

镜像队列包括一个主队列和多个镜像队列。主与镜像之间的消息是异步方式

消息发送给主队列，随后会复制到所有的镜像队列；当主队列所在的节点发生故障，镜像可以接管主队列

配置方法

在某一节点的 Admin-policies中配置镜像策略，结果如下：
在这里插入图片描述

参数配置：

其中，Name为策略名称；
Pattern为通过正则将部分队列设为镜像策略的队列。
- ^ 是将所有的队列设为镜像队列。
Definition 镜像定义。包括 ha-mode,ha-params,ha-sync-mode
- ha-mode 镜像队列模式：
  - all 在集群的所有代理（节点）上进行镜像。
  - exactly 指定个数的节点上进行镜像。个数由 ha-params指定。建议设置的副本值为大多数节点N / 2 + 1
    - ha-params = 1 表示只有一个主节点
    - ha-params = 3 需要2个节点作为镜像节点（一主二从）
  - nodes 在指定节点进行镜像。节点名称通过 ha-params指定。
- ha-params ha-mode 需要的参数
- ha-sync-mode 镜像队列间的同步方式。
  - automatic 自动将主节点消息同步到镜像节点。 – 新加入group的节点会自动同步消息
  - manually 手动操作完成消息同步。镜像队列只接收新消息，已被发送的消息无法备份，。如果主队列在所有未同步的消息耗尽之前宕机，则这些消息将丢失。
Priority Priority: 可选参数， policy的优先级。

显示的 mirror-policy 为该队列应用的镜像策略。+n 表示同步副本为n个

新镜像同步策略

ha-sync-mode	说明
manual	这是默认模式。新队列镜像将不接收现有消息，它只接收新消息。一旦使用者耗尽了仅存在于主服务器上的消息，新的队列镜像将随着时间的推移成为主服务器的精确副本。如果主队列在所有未同步的消息耗尽之前失败，则这些消息将丢失。您可以手动完全同步队列，详情请参阅未同步的镜像部分。
automatic	当新镜像加入时，队列将自动同步。值得重申的是，队列同步是一个阻塞操作。如果队列很小，或者您在RabbitMQ节点和ha-sync-batch-size之间有一个快速的网络，那么这是一个很好的选择。

从节点晋升策略

当主节点出现故障，最老的从节点会提升为新的主节点。

参数：

ha-promote-on-shutdown，ha-promote-on-failure

ha-promote-on-shutdown/ha-promote-on-failure	说明
when-synced	从节点与主节点完成数据同步，才会被提升为主节点
always	无论什么情况下从节点都将被提升为主节点

Haproxy+Keepalive 实现高可用负载均衡

(67条消息) 学会Haproxy+Keepalived实现高可用负载均衡项目搭建一片就够了！！！_仇亚峰的博客-CSDN博客

Haproxy，LVS，Nginx

负载均衡工具。

不同的适用场景：

中小型Web应用。比如日 PV 小于1000万，用 Nginx 就完全可以。
1. 如果机器不少，可以用 DNS 轮询，LVS 所耗费的机器还是比较多的
大型网站或重要的服务，且服务器比较多时，可以考虑用 LVS。

Web前端采用Nginx/Haproxy + Keepalived 负载均衡；后端 MySQL数据库一主多从和读写分离。采用LVS+Keepalived 架构。

LVS

优点：

抗负载能力强、是工作在传输层上仅作分发之用，没有流量的产生，这个特点也决定了它在负载均衡软件里的性能最强的，对内存和 cpu资源消耗比较低。
配置性比较低，这是一个缺点也是一个优点，因为没有可太多配置的东西，所以并不需要太多接触，大大减少了人为出错的几率。
工作稳定，因为其本身抗负载能力很强，自身有完整的双机热备方案，如 LVS + Keepalived。
无流量，LVS 只分发请求，而流量并不从它本身出去，这点保证了均衡器 IO 的性能不会受到大流量的影响。
应用范围比较广，因为 LVS 工作在传输层，所以它几乎可以对所有应用做负载均衡，包括 http、数据库、在线聊天室等等。

LVS 的缺点

软件本身不支持正则表达式处理，不能做动静分离；而现在许多网站在这方面都有较强的需求，这个是 Nginx、HAProxy + Keepalived 的优势所在。
如果是网站应用比较庞大的话，LVS/DR + Keepalived 实施起来就比较复杂了，相对而言，Nginx / HAProxy + Keepalived 就简单多了。

Nginx

Nginx 的优点

跨平台：Nginx 可以在大多数 Unix like OS编译运行，而且也有 Windows 的移植版本
配置异常简单：非常容易上手。配置风格跟程序开发一样，神一般的配置
非阻塞、高并发连接：官方测试能够支撑5万并发连接，在实际生产环境中跑到2～3万并发连接数
事件驱动：通信机制采用 epoll 模型，支持更大的并发连接
Master/Worker 结构：一个 master 进程，生成一个或多个 worker 进程
内存消耗小：处理大并发的请求内存消耗非常小。在3万并发连接下，开启的10个 Nginx 进程才消耗150M 内存（15M*10=150M）
内置的健康检查功能：如果 Nginx 代理的后端的某台 Web 服务器宕机了，不会影响前端访问
节省带宽：支持 GZIP 压缩，可以添加浏览器本地缓存的 Header 头
稳定性高：用于反向代理，宕机的概率微乎其微

Nginx 的缺点

Nginx 仅能支持http、https 和 Email 协议，适用范围小。
对后端服务器的健康检查，只支持通过端口来检测，不支持通过 url 来检测。不支持 Session 的直接保持，但能通过 ip_hash 来解决。

Haproxy

HAProxy 支持两种代理模式 TCP（四层）和HTTP（七层），也是支持虚拟主机的。
HAProxy 的优点能够补充 Nginx 的一些缺点，比如支持 Session 的保持，Cookie 的引导；同时支持通过获取指定的 url 来检测后端服务器的状态。
HAProxy 跟 LVS 类似，本身就只是一款负载均衡软件；单纯从效率上来讲 HAProxy 会比 Nginx 有更出色的负载均衡速度，在并发处理上也是优于 Nginx 的。
HAProxy 支持 TCP 协议的负载均衡转发，可以对 MySQL 读进行负载均衡，对后端的 MySQL 节点进行检测和负载均衡，大家可以用 LVS+Keepalived 对 MySQL 主从做负载均衡。
HAProxy 负载均衡策略非常多：Round-robin（轮循）、Weight-round-robin（带权轮循）、source（原地址保持）、RI（请求URL）、rdp-cookie（根据cookie）。

HAPorxy缺点：

（1）不支持POP/SMTP协议
（2）不支持SPDY协议
（3）不支持HTTP cache功能。现在不少开源的lb项目，都或多或少具备HTTP cache功能。
（4）重载配置的功能需要重启进程，虽然也是soft restart，但没有Nginx的reaload更为平滑和友好。
（5）多进程模式支持不够好

Keepalived

高可用——Keepalived安装部署使用详解 - 掘金

Keepalived 的优势所在。

如果是网站应用比较庞大的话，LVS/DR + Keepalived 实施起来就比较复杂了，相对而言，Nginx / HAProxy + Keepalived 就简单多了。

Nginx

Nginx 的优点

跨平台：Nginx 可以在大多数 Unix like OS编译运行，而且也有 Windows 的移植版本
配置异常简单：非常容易上手。配置风格跟程序开发一样，神一般的配置
非阻塞、高并发连接：官方测试能够支撑5万并发连接，在实际生产环境中跑到2～3万并发连接数
事件驱动：通信机制采用 epoll 模型，支持更大的并发连接
Master/Worker 结构：一个 master 进程，生成一个或多个 worker 进程
内存消耗小：处理大并发的请求内存消耗非常小。在3万并发连接下，开启的10个 Nginx 进程才消耗150M 内存（15M*10=150M）
内置的健康检查功能：如果 Nginx 代理的后端的某台 Web 服务器宕机了，不会影响前端访问
节省带宽：支持 GZIP 压缩，可以添加浏览器本地缓存的 Header 头
稳定性高：用于反向代理，宕机的概率微乎其微

Nginx 的缺点

Nginx 仅能支持http、https 和 Email 协议，适用范围小。
对后端服务器的健康检查，只支持通过端口来检测，不支持通过 url 来检测。不支持 Session 的直接保持，但能通过 ip_hash 来解决。

Haproxy

HAProxy 支持两种代理模式 TCP（四层）和HTTP（七层），也是支持虚拟主机的。
HAProxy 的优点能够补充 Nginx 的一些缺点，比如支持 Session 的保持，Cookie 的引导；同时支持通过获取指定的 url 来检测后端服务器的状态。
HAProxy 跟 LVS 类似，本身就只是一款负载均衡软件；单纯从效率上来讲 HAProxy 会比 Nginx 有更出色的负载均衡速度，在并发处理上也是优于 Nginx 的。
HAProxy 支持 TCP 协议的负载均衡转发，可以对 MySQL 读进行负载均衡，对后端的 MySQL 节点进行检测和负载均衡，大家可以用 LVS+Keepalived 对 MySQL 主从做负载均衡。
HAProxy 负载均衡策略非常多：Round-robin（轮循）、Weight-round-robin（带权轮循）、source（原地址保持）、RI（请求URL）、rdp-cookie（根据cookie）。

HAPorxy缺点：

相关阅读:
【pytest】标记冒烟用例 @pytest.mark.smoke
LeeCode 1787 DP
Spring 面试题及答案整理，最新面试题
 性价比高一点的蓝牙耳机有哪几款？高性价比蓝牙耳机推荐
 docker部署excalidraw画图工具(银角大王课堂使用的画图软件搭建)
influxdb 2.*版本与1.*版本区别
 PVT v2: Improved Baselines with Pyramid Vision Transformer
Centos7安装Nginx
计算机毕业设计之java+ssm服装定制系统
 Nginx 单个端口代理Minio
原文地址：https://blog.csdn.net/qq_45888242/article/details/133586841

RabbitMQ学习笔记

RabbitMQ

中间件

保证高可靠：

持久化：

消息分发策略 机制

安装

消息模型

RabbitMQ优缺点

应用场景

Rabbitmq架构

常见端口

常用命令：

Linux

windows

MQ消息投递、ACK机制总结

RabbitMQ的消息模型

简单模式

工作模式

消息应答（消费者确认）

自动应答

手动确认

Springboot 消费者确认：

队列持久化

消息持久化

发布确认（生产者确认）

订阅发布

交换机类型：

死信队列：

应用场景

延迟队列

设置TTL

消息设置：

队列设置：

Delayed Message 插件

延时队列和死信队列

发布确认高级（生产者确认）

confirmCallback

returnedMessage

备份交换机

幂等性

重复消费：

优先级队列

惰性队列

镜像队列

配置方法

新镜像同步策略

从节点晋升策略

Haproxy+Keepalive 实现高可用负载均衡

Haproxy，LVS，Nginx

LVS

Nginx

Haproxy

Keepalived

Nginx

Haproxy

消息分发策略机制