1 服务异步通讯

RabbitMQ高级特性

1.1 消息可靠性

消息丢失：

· 发送时丢失：· 生产者发送的消息未送达exchange· 消息到达exchange后未到达queue
· MQ宕机，queue将消息丢失
· consumer接收到消息后未消费就宕机

如何确保RabbitMQ消息的可靠性？

· 对于生产者，开启生产者确认机制，确保生产者的消息能到达队列
· 对于MQ，开启持久化功能，确保消息未消费前在队列中不会丢失
· 对于消费者，开启消费者确认机制为auto，由spring确认消息处理成功后完成ack，
· 对于无限的重试，开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理

1.1.1 生产者消息确认

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。

结果有两种请求：

· publisher-confirm，发送者确认· 消息成功投递到交换机，返回ack· 消息未投递到交换机，返回nack
· publisher-return，发送者回执· 消息投递到交换机，但没有路由到队列。返回ack，及路由失败原因

！！！ 确认机制发送消息时，需要给每个消息设置一个全局唯一id，以区分不同消息，避免ack冲突

===============================================

SpringAMQP实现生产者确认
1、在publisher的application.yml中添加配置：

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true

配置说明：

· publisher-confirm-type：开启publisher-confirm，这里支持两种类型：· simple：同步等待confirm结果，直到超时· correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
· publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
· template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false，则直接丢弃消息

2、每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置：

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {//获取RabbitTemplate对象RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);//配置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {//记录日志log.error("消息发送到队列失败，响应码：{},失败原因：{},交换机：{},路由key：{},消息：{}",replyCode,replyText,exchange,routingKey,message);//如果有需要，重发消息});}
}

3、发送消息，指定消息ID、消息ConfirmCallback
不需要全局配置，只是在发消息那一刻

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendMessage2SimpleQueue() throws InterruptedException {String routingKey = "simple";// 准备消息String message = "hello, spring amqp!";//准备CorrelationDataCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());//准备ConfirmCallbackcorrelationData.getFuture().addCallback(confirm -> {//判断结果if (confirm.isAck()){//ACKlog.debug("消息投递到交换机 成功！ 消息id：{}",correlationData.getId());}else {//NACKlog.error("消息投递到交换机 失败！ 消息id：{}",correlationData.getId());}}, throwable -> {//失败  记录日志log.error("消息发送失败！",throwable);//重发消息});//发送消息rabbitTemplate.convertAndSend("amq.topic", routingKey, message,correlationData);}
}

1.1.2 消息持久化

MQ默认是内存存储消息，开启持久化功能可确保缓存在MQ中的消息不丢失 。
1、交换机持久化：

2、队列持久化：

3、消息持久化，发送持久消息
SpringAMQP中的消息默认是持久的，可通过MessageProperties中的DeliveryMode来指定：

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testDurableMessage(){//1 准备消息Message message = MessageBuilder.withBody("hello,spirng".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();//2 发送消息rabbitTemplate.convertAndSend("simple.queue",message);}}

1.1.3 消费者消息确认

RabbitMQ支持消费者确认机制，即：消费者处理消息后可向MQ发送ack回执，MQ收到ack回执后才会删除该消息。

SpringAMQP允许配置三种确认模式：

· manual：手动ack，需要在业务代码结束后，调用api发送ack
· auto：自动ack，由spring监测listener代码是否出现异常，无异常则返回ack；抛出异常则返回nack
· none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

配置方式是修改application.yml文件:

spring:rabbitmq:listener:simple:prefetch: 1acknowledge-mode: auto # manual # none

1.1.4 消费失败重试机制

当消费者出现异常后，消息会不断requeue（重新入队）到队列，再异常再次重复，无限循环，导致mq的消息处理飙升

可利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列

spring:rabbitmq:listener:simple:prefetch: 1retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初始的失败等待时长为1秒multiplier: 1 # 下次失败的等待时长倍数 ，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

消费者失败消息处理策略
在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口处理
包含三种实现：

· RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。（默认方式）
· ImmediateRequestMessageRecoverer：重试耗尽后，返回nack，消息重新入队
· RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

对于第三种，如下图示：

测试RepublishMessageRecoverer：

@Configuration
public class ErrorMessageConfig {/****  测试RepublishMessageRecoverer处理模式*///定义接收失败消息的交换机、队列、绑定关系:@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue");}@Beanpublic Binding errorBinding(){return BindingBuilder.bind(errorQueue()).to(errorMessageExchange()).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}
}

1.2 死信交换机

死信（dead letter）：

· 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false
· 消息是一个过期消息，超时无人消费
· 要投递的队列消息堆积满了，最早的消息可能成为死信

如果该队列配置了dead-letter-exchange属性，指定了一个交换机，这个交换机称为 死信交换机 （Dead Letter Exchange，DLX）

1.2.1 TTL

Time-To-Live，如果一个队列中的消息TTL结束仍未消费，则会变为死信，ttl超时分为两种情况：

· 消息所在的队列设置了存活时间
· 消息本身设置了存活时间

给消息所在队列设置存活时间

配置类ttlMessageConfig：

@Configuration
public class TTLMessageConfig {@Beanpublic DirectExchange ttlDirectExchange(){return new DirectExchange("ttl.direct");}@Beanpublic Queue ttlQueue(){return QueueBuilder.durable("ttl.queue").ttl(10000).deadLetterExchange("dl.direct").deadLetterRoutingKey("dl").build();}@Beanpublic Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");}}

消费者监听类SpringRabbitListener：

@Slf4j
@Component
public class SpringRabbitListener {@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dl.queue",durable = "true"),exchange = @Exchange(name = "dl.direct"),key = "dl"))public void listenDlQueue(String msg){log.info("消费者接收到了dl.queue的延迟消息");}}

消费本身设置存活时间

 @Testpublic void testTTLMessage(){//1 准备消息Message message = MessageBuilder.withBody("hello,ttl message ".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).setExpiration("5000").build();//2 发送消息rabbitTemplate.convertAndSend("ttl.direct","ttl",message);//记录日志log.info("消息已经成功发送！！");}

在两者都设置时间的情况下，以较短的那一个为准

1.2.2 延迟队列

利用TTL结合死信交换机，这种消息模式称为延迟队列（Delay Queue）
延迟队列使用场景包括：

· 延迟发送短信
· 用户下单，如果用户在15分钟内未支付，则自动取消
· 预约工作会议，20分钟后自动通知所有参会人员 

安装DelayExchange插件

安装MQ时将数据卷挂载到mq-plugins这个文件夹中

docker run \-e RABBITMQ_DEFAULT_USER=itcast \-e RABBITMQ_DEFAULT_PASS=123321 \-v mq-plugins:/plugins \--name mq \--hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3.8-management

1.下载插件
RabbitMQ有一个官方的插件社区，地址为：https://www.rabbitmq.com/community-plugins.html

我采用的是3.9.0版本，地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

2.上传插件
我们之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我们使用下面命令查看数据卷：

docker volume inspect mq-plugins

可以得到下面结果：

然后将插件上传到该目录：

3.安装插件
需要进入MQ容器内部来执行安装。我的容器名为mq，所以执行下面命令：

docker exec -it mq bash

执行时，请将其中的 -it 后面的mq替换为你自己的容器名.

进入容器内部后，执行下面命令开启插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

结果如下：

4.使用插件
DelayExchange插件是对官方原生的Exchange做了功能的升级：

· 将DelayExchange接受到的消息暂存在内存中
· 在DelayExchange中计时，超时后才投递消息到队列中

创建交换机

发消息需要加头信息

SpringAMQP使用延迟队列插件
DelayExchange本质还是官方三种交换机，只是添加延迟功能。
因此使用时只需声明一个交换机，交换机的类型可以是任意类型，然后设定delayed属性为true

基于注解形式：

基于java代码方式：

在为delay为true的交换机中发送消息，一定要给消息添加一个header：x-delay，值为延迟时间，单位为毫秒：

1.3 惰性队列

1.3.1 消息堆积

当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。最早接收的消息就可能成为死信，会被丢弃，即消息堆积

解决思路：

· 增加更多消费者，提噶奥消费速度
· 在消费者内开启线程池加快消息处理速度（适合消息处理业务耗时长）
· 扩大队列容积，提高堆积上限

1.3.2 惰性队列

特征：

· 接收到消息后直接存入磁盘而非内存
· 消费者要消费消息是才会从磁盘中读取并加载到内存
· 支持数百万条消息存储

方式一：
设置一个队列为惰性队列，只需在声明队列时，指定x-queue-mode属性为lazy即可。
可以通过命令行将一个运行中的队列修改为惰性队列：

方式二：
用SpringAMQP声明惰性队列
1）@Bean方式

2）注销方式

惰性队列优点：

· 基于磁盘存储，消息上限高
· 没有间歇性的page-out，性能比较稳定

惰性队列缺点：

· 基于磁盘存储，消息时效性降低
· 性能受限于磁盘IO

1.4 MQ集群

1.4.1 集群分类

RabbitMQ是基于Erlang语言编写，Erlang是一个面向并发的语言，天然支持集群模式。

集群有两种模式：

· 普通集群：一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群并发能力（当集群中一个节点出现故障，则节点上的所有消息、队列都会丢失）
· 镜像集群：一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性

RabbitMQ3.8版本后，仲裁队列代替镜像集群，底层采用Raft协议确保主从的数据一致性

1.4.2 普通集群

标准集群（classic cluster）
特征：

· 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包括队列中的消息。（队列是不共享的，其他节点是存有当前节点队列的描述信息，相当于引用）
· 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回  
· 队列所在节点宕机，队列中的消息就会丢失

1.4.2.1 集群分类

计划部署3节点的mq集群：

集群中的节点标示默认都是：rabbit@[hostname]，因此以上三个节点的名称分别为：

• rabbit@mq1
• rabbit@mq2
• rabbit@mq3

1.4.2.2 获取Cookie

RabbitMQ底层依赖于Erlang，Erlang虚拟机就是一个面向分布式的语言，默认就支持集群模式。集群模式中的每个RabbitMQ 节点使用 cookie 来确定它们是否被允许相互通信。

要使两个节点能够通信，它们必须具有相同的共享秘密，称为Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。

每个集群节点必须具有相同的 cookie。实例之间也需要它来相互通信。

在mq容器中获取一个cookie值，作为集群的cookie。执行下面的命令：

docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie

可以看到cookie值如下：

KQHDBDGOBZLFIVDITWMW

接下来，停止并删除当前的mq容器，重新搭建集群。

docker rm -f mq

1.4.2.3 集群配置

在/tmp目录新建一个配置文件 rabbitmq.conf：

cd /tmp
# 创建文件
touch rabbitmq.conf

文件内容如下：

loopback_users.guest = false
listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@mq1
cluster_formation.classic_config.nodes.2 = rabbit@mq2
cluster_formation.classic_config.nodes.3 = rabbit@mq3

再创建一个文件，记录cookie

cd /tmp
# 创建cookie文件
touch .erlang.cookie
# 写入cookie
echo "KQHDBDGOBZLFIVDITWMW" > .erlang.cookie
# 修改cookie文件的权限
chmod 600 .erlang.cookie

准备三个目录,mq1、mq2、mq3：

cd /tmp
# 创建目录
mkdir mq1 mq2 mq3

然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3：

# 进入/tmp
cd /tmp
# 拷贝
cp rabbitmq.conf mq1
cp rabbitmq.conf mq2
cp rabbitmq.conf mq3
cp .erlang.cookie mq1
cp .erlang.cookie mq2
cp .erlang.cookie mq3

1.4.2.4 启动集群

创建一个网络：

docker network create mq-net

docker volume create

运行命令

docker run -d --net mq-net \
-v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq1 \
--hostname mq1 \
-p 8071:5672 \
-p 8081:15672 \
rabbitmq:3.8-management

docker run -d --net mq-net \
-v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq2 \
--hostname mq2 \
-p 8072:5672 \
-p 8082:15672 \
rabbitmq:3.8-management

docker run -d --net mq-net \
-v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq3 \
--hostname mq3 \
-p 8073:5672 \
-p 8083:15672 \
rabbitmq:3.8-management

1.4.3 镜像集群

本质是主从模式，具备下面特征：

· 交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份
· 创建队列的节点被称为该队列的主节点，备份到的其它节点叫做该队列的镜像节点
· 一个队列的主节点可能是另一个队列的镜像节点
· 所有操作都是主节点完成，然后同步给镜像节点
· 主节点宕机后，镜像节点替换为新的主节点

当主节点接收到消费者的ACK时，所有镜像都会删除节点中的数据。

• 镜像队列结构是一主多从（从就是镜像）
• 所有操作都是主节点完成，然后同步给镜像节点
• 主宕机后，镜像节点会替代成新的主（如果在主从同步完成前，主就已经宕机，可能出现数据丢失）
• 不具备负载均衡功能，因为所有操作都会有主节点完成（但是不同队列，其主节点可以不同，可以利用这个提高吞吐量）

镜像模式的配置有3种模式：

1.4.3.1 exactly模式

rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

• rabbitmqctl set_policy：固定写法
• ha-two：策略名称，自定义
• "^two\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以two.开头的队列名称
• '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 策略内容
  • "ha-mode":"exactly"：策略模式，此处是exactly模式，指定副本数量
  • "ha-params":2：策略参数，这里是2，就是副本数量为2，1主1镜像
  • "ha-sync-mode":"automatic"：同步策略，默认是manual，即新加入的镜像节点不会同步旧的消息。如果设置为automatic，则新加入的镜像节点会把主节点中所有消息都同步，会带来额外的网络开销

1.4.3.2 all模式

rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}'

• ha-all：策略名称，自定义
• "^all\."：匹配所有以all.开头的队列名
• '{"ha-mode":"all"}'：策略内容
  • "ha-mode":"all"：策略模式，此处是all模式，即所有节点都会称为镜像节点

1.4.3.3 nodes模式

rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

• rabbitmqctl set_policy：固定写法
• ha-nodes：策略名称，自定义
• "^nodes\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以nodes.开头的队列名称
• '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 策略内容
  • "ha-mode":"nodes"：策略模式，此处是nodes模式
  • "ha-params":["rabbit@mq1", "rabbit@mq2"]：策略参数，这里指定副本所在节点名称

1.4.4 仲裁集群

3.8版本以后才有的新功能，用来替代镜像队列，具备特征：

· 与镜像队列一样，都是主从模式，支持主从数据同步
· 使用非常简单，无复杂的配置
· 主从同步基于Raft协议，强一致 

1.添加仲裁队列

在任意控制台添加一个队列，一定要选择队列类型为Quorum类型。

SpringAMQP创建仲裁队列：

SpringAMQP连接集群，只需在yaml中配置即可：

特殊1：:集群扩容

加入集群

1）启动一个新的MQ容器：

docker run -d --net mq-net \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq4 \
--hostname mq5 \
-p 8074:15672 \
-p 8084:15672 \
rabbitmq:3.8-management

2）进入容器控制台：

docker exec -it mq4 bash

3）停止mq进程

rabbitmqctl stop_app

4）重置RabbitMQ中的数据：

rabbitmqctl reset

5）加入mq1：

rabbitmqctl join_cluster rabbit@mq1

6）再次启动mq进程

rabbitmqctl start_app

增加仲裁队列副本

先查看下quorum.queue这个队列目前的副本情况，进入mq1容器：

docker exec -it mq1 bash

执行命令：

rabbitmq-queues quorum_status "quorum.queue"

结果：

现在，我们让mq4也加入进来：

rabbitmq-queues add_member "quorum.queue" "rabbit@mq4"

结果：

再次查看：

rabbitmq-queues quorum_status "quorum.queue"

查看控制台，发现quorum.queue的镜像数量也从原来的 +2 变成了 +3：

本文链接：https://my.lmcjl.com/post/13319.html