12.RabbitMQ消息Confirm模式
12.1 Confirm模式简介
消息的confirm确认机制，是指生产者投递消息后，到达了消息服务器Broker里面的exchange交换机，则会给生产者一个应答，生产者接收到应答，用来确定这条消息是否正常的发送到Broker的exchange中，这也是消息可靠性投递的重要保障；
12.2 具体代码设置
| 1 配置文件application.yml 开启确认模式：spring.rabbitmq.publisher-confirm-type=_correlated_2 写一个类实现implements RabbitTemplate.ConfirmCallback，判断成功和失败的ack结果，可以根据具体的结果，如果ack为false，对消息进行重新发送或记录日志等处理；设置rabbitTemplate的确认回调方法3 rabbitTemplate.setConfirmCallback(messageConfirmCallBack); |
|---|
参考代码：
| @Component
public class MessageConfirmCallBack implements RabbitTemplate.ConfirmCallback {
/**
* 交换机收到消息后，会回调该方法
*
* @param correlationData 相关联的数据
* @param ack 有两个取值，true和false，true表示成功：消息正确地到达交换机，反之false就是消息没有正确地到达交换机
* @param cause 消息没有正确地到达交换机的原因是什么
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
if (ack) {
//正常
} else {
//不正常的，可能需要记日志或重新发送
}
}
| } |
|---|
发消息参考代码
| @Service
public class MessageService {
@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private MessageConfirmCallBack messageConfirmCallBack;
@PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
public void init() {
rabbitTemplate.setConfirmCallback(messageConfirmCallBack);
}
/**
* 发送消息
*/
public void sendMessage() {
//关联数据对象
CorrelationData correlationData = new CorrelationData();
correlationData.setId(“O159899323”); //比如设置一个订单ID，到时候在confirm回调里面，你就可以知道是哪个订单没有发送到交换机上去
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE + 123, “info”, “hello”, correlationData);
System.out.println(“消息发送完毕…”);
}
}
|
| — |
两个示例：可以在发消息时直接实现接口
13. RabbitMQ消息Return模式
rabbitmq 整个消息投递的路径为：
producer —> exchange —> queue —> consumer
消息从 producer 到 exchange 则会返回一个 confirmCallback；
消息从 exchange –> queue 投递失败则会返回一个 returnCallback；
我们可以利用这两个callback控制消息的可靠性投递；
开启 确认模式；
使用rabbitTemplate.setConfirmCallback设置回调函数，当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理；
注意配置文件中，开启 退回模式；
| spring.rabbitmq.publisher-returns: true |
|---|
使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到
queue失败后，则会将消息退回给producer，并执行回调函数returnedMessage；
|
@Component
public class MessageReturnCallBack implements RabbitTemplate.ReturnsCallback {
/**
* 当消息从交换机 没有正确地 到达队列，则会触发该方法
* 如果消息从交换机 正确地 到达队列了，那么就不会触发该方法
*
* @param returned
*/
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.println(“消息return模式：” + returned);
}
| } |
|---|
参考发送代码
| @Service
public class MessageService {
@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private MessageReturnCallBack messageReturnCallBack;
@PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
public void init() {
rabbitTemplate.setReturnsCallback(messageReturnCallBack);
}
/**
* 发送消息
*/
public void sendMessage() {
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, “info123”, “hello”);
System.out.println(“消息发送完毕…”);
}
}
|
| — |
代码二，发送消息时直接实现RabbitTemplate.ReturnsCallback
| @Service
public class MessageService implements RabbitTemplate.ReturnsCallback {
@Resource
private RabbitTemplate rabbitTemplate;
@PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
public void init() {
rabbitTemplate.setReturnsCallback(this);
}
/**
* 当消息从交换机 没有正确地 到达队列，则会触发该方法
* 如果消息从交换机 正确地 到达队列了，那么就不会触发该方法
*
* @param returned
/
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.println(“消息return模式：” + returned);
}
/*
* 发送消息
*/
public void sendMessage() {
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, “info123”, “hello”);
System.out.println(“消息发送完毕…”);
}
}
|
| — |
代码参考3 同时实现RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback
|
@Service
public class MessageService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
@Resource
private RabbitTemplate rabbitTemplate;
@PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/**
* 交换机收到消息后，会回调该方法
*
* @param correlationData 相关联的数据
* @param ack 有两个取值，true和false，true表示成功：消息正确地到达交换机，反之false就是消息没有正确地到达交换机
* @param cause 消息没有正确地到达交换机的原因是什么
/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
if (ack) {
//正常
} else {
//不正常的
}
}
/*
* 当消息从交换机 没有正确地 到达队列，则会触发该方法
* 如果消息从交换机 正确地 到达队列了，那么就不会触发该方法
*
* @param returned
/
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.println(“消息return模式：” + returned);
}
/*
* 发送消息
*/
public void sendMessage() {
//关联数据对象
CorrelationData correlationData = new CorrelationData();
correlationData.setId(“O159899323”); //比如设置一个订单ID，到时候在confirm回调里面，你就可以知道是哪个订单没有发送到交换机上去
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, “info123”, “hello”, correlationData);
System.out.println(“消息发送完毕…”);
}
}
|
| — |
14. RabbitMQ交换机详细属性
14.1具体参数
1、Name：交换机名称；就是一个字符串
2、Type：交换机类型，direct, topic, fanout, headers四种
3、Durability：持久化，声明交换机是否持久化，代表交换机在服务器重启后是否还存在；
4、Auto delete：是否自动删除，曾经有队列绑定到该交换机，后来解绑了，那就会自动删除该交换机；
5、Internal：内部使用的，如果是yes，客户端无法直接发消息到此交换机，它只能用于交换机与交换机的绑定。
6、Arguments：只有一个取值alternate-exchange，表示备用交换机；
14.2代码演示
结论1：没发消息之前不会创建交换机和对列
结论2：发消息后，如果交换机不存在，才开始创建交换机，如果队列不存在，则创建新的对列
结论3：创建交换机或者队列完成后再重新创建，如果修改交换机或队列参数则会报错
406错误（inequivalent arg ‘durable’ for exchange ‘exchange.durability’ in vhost ‘powernode’: received ‘false’ but current is ‘true’, class-id=40, method-id=10)）
结论4：设置持久化为false ，重启rabbitmq-server，则交换机丢失，实验durable参数，先看下控制台，然后重启rabbitmq-server
结论5：实验自动删除为 true ，从控制台上手动解绑，会发现自动删除
14.3 备用交换机
14.3.1 备用交换机使用场景
当消息经过交换器准备路由给队列的时候，发现没有对应的队列可以投递信息，在rabbitmq中会默认丢弃消息，如果我们想要监测哪些消息被投递到没有对应的队列，我们可以用备用交换机来实现，可以接收备用交换机的消息，然后记录日志或发送报警信息。
14.3.2 主要代码和注意事项
备用交换机示例如下：
注意：备用交换机一般使用fanout交换机
测试时：指定一个错误路由
重点：普通交换机设置参数绑定到备用交换机
| Map<String, Object> arguments = new HashMap<>();
//指定当前正常的交换机的备用交换机是谁
arguments.put(“alternate-exchange”, EXCHANGE_ALTERNATE);
//DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
return new DirectExchange(EXCHANGE, true, false, arguments);
| //return ExchangeBuilder.directExchange(EXCHANGE).withArguments(args).build(); |
|---|
14.3.3 参考配置代码
| @Configuration
public class RabbitConfig {
//交换机的名字，就是一个字符串
public static final String EXCHANGE = “exchange”;
//队列的名字，就是一个字符串
public static final String QUEUE = “queue”;
//定义的一个路由键
public static final String INFO = “info”;
//-------------------------------------------
//交换机的名字，就是一个字符串
public static final String EXCHANGE_ALTERNATE = “exchange.alternate”;
//队列的名字，就是一个字符串
public static final String QUEUE_ALTERNATE = “queue.alternate”;
//定义的一个路由键
public static final String ALTERNATE = “alternate”;
@Bean
public DirectExchange directExchange() {
Map<String, Object> arguments = new HashMap<>();
arguments.put(“alternate-exchange”, EXCHANGE_ALTERNATE); //指定当前正常的交换机的备用交换机是谁
//DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
return new DirectExchange(EXCHANGE, true, false, arguments);
}
/**
* 声明一个队列
*
* @return
/
@Bean
public Queue queue() {
return QueueBuilder.durable(QUEUE).build();
}
/*
* @Qualifier 限定bean的名字是 directExchange 的Bean
*
* @param directExchange
* @return
/
@Bean
public Binding binding(DirectExchange directExchange, Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with(INFO);
}
//-----------------------------------------
/*
* 备用交换机需要用Fanout交换机；
*
* @return
*/
@Bean
public FanoutExchange alternateExchange() {
//DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
return new FanoutExchange(EXCHANGE_ALTERNATE, true, false);
}
@Bean
public Queue alternateQueue() {
return QueueBuilder.durable(QUEUE_ALTERNATE).build();
}
@Bean
public Binding alternateBnding(FanoutExchange alternateExchange, Queue alternateQueue) {
return BindingBuilder.bind(alternateQueue).to(alternateExchange);
}
}
|
| — |
14.3.4 参考发送消息代码
| @Service
public class MessageService {
@Resource
private RabbitTemplate rabbitTemplate;
/**
* 发送消息
*/
public void sendMessage() {
//我们故意写错路由key,由于我们正常交换机设置了备用交换机，所以该消息就会进入备用交换机 //从而进入备用对列，我们可以写一个程序接收备用对列的消息，接收到后通知相关人员进行处理 //如果正常交换机没有设置备用交换机，则该消息会被抛弃。
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, “info1223”, “hello”);
System.out.println(“消息发送完毕…”);
}
}
|
| — |
15. RabbitMQ队列详细属性
15.1 具体参数
Type：队列类型
Name：队列名称，就是一个字符串，随便一个字符串就可以；
Durability：声明队列是否持久化，代表队列在服务器重启后是否还存在；
Auto delete： 是否自动删除，如果为true，当没有消费者连接到这个队列的时候，队列会自动删除；
Exclusive：exclusive属性的队列只对首次声明它的连接可见，并且在连接断开时自动删除；
基本上不设置它，设置成false
Arguments：队列的其他属性，例如指定DLX（死信交换机等）；
1、x-expires：Number
当Queue（队列）在指定的时间未被访问，则队列将被自动删除；
2、x-message-ttl：Number
发布的消息在队列中存在多长时间后被取消（单位毫秒）；
3、x-overflow：String
设置队列溢出行为，当达到队列的最大长度时，消息会发生什么，有效值为Drop Head或Reject Publish；
4、x-max-length：Number
队列所能容下消息的最大长度，当超出长度后，新消息将会覆盖最前面的消息，类似于Redis的LRU算法；
5、 x-single-active-consumer：默认为false
激活单一的消费者，也就是该队列只能有一个消息者消费消息；
6、x-max-length-bytes：Number
限定队列的最大占用空间，当超出后也使用类似于Redis的LRU算法；
7、x-dead-letter-exchange：String
指定队列关联的死信交换机，有时候我们希望当队列的消息达到上限后溢出的消息不会被删除掉，而是走到另一个队列中保存起来；
8.x-dead-letter-routing-key：String
指定死信交换机的路由键，一般和6一起定义；
9.x-max-priority：Number
如果将一个队列加上优先级参数，那么该队列为优先级队列；
（1）、给队列加上优先级参数使其成为优先级队列
x-max-priority=10【0-255取值范围】
（2）、给消息加上优先级属性
通过优先级特性，将一个队列实现插队消费；
| MessageProperties messageProperties=new MessageProperties();
| messageProperties.setPriority(8); |
|---|
10、x-queue-mode：String（理解下即可）
队列类型x-queue-mode=lazy懒队列，在磁盘上尽可能多地保留消息以减少RAM使用，如果未设置，则队列将保留内存缓存以尽可能快地传递消息；
11、x-queue-master-locator：String（用的较少，不讲）
在集群模式下设置队列分配到的主节点位置信息；
每个queue都有一个master节点，所有对于queue的操作都是事先在master上完成，之后再slave上进行相同的操作；
每个不同的queue可以坐落在不同的集群节点上，这些queue如果配置了镜像队列，那么会有1个master和多个slave。
基本上所有的操作都落在master上，那么如果这些queues的master都落在个别的服务节点上，而其他的节点又很空闲，这样就无法做到负载均衡，那么势必会影响性能；
关于master queue host 的分配有几种策略，可以在queue声明的时候使用x-queue-master-locator参数，或者在policy上设置queue-master-locator，或者直接在rabbitmq的配置文件中定义queue_master_locator，有三种可供选择的策略：
（1）min-masters：选择master queue数最少的那个服务节点host；
（2）client-local：选择与client相连接的那个服务节点host；
（3）random：随机分配；
15.2 参考代码
|
@Configuration
public class RabbitConfig {
public static final String EXCHANGE = “exchange”;
public static final String QUEUE = “queue”;
public static final String KEY = “info”;
QueueBuilder builder;
@Bean
public DirectExchange directExchange() {
return ExchangeBuilder.directExchange(EXCHANGE).build();
}
@Bean
public Queue queue() {
Map<String, Object> arguments = new HashMap<>();
//arguments.put(“x-expires”, 5000);
//arguments.put(“x-max-length”, 5);
//arguments.put(“x-overflow”, “reject-publish”);
arguments.put(“x-single-active-consumer”, false); //TODO ???
//arguments.put(“x-max-length-bytes”, 20); // 单位是字节
//arguments.put(“x-max-priority”, 10); // 0-255 //表示把当前声明的这个队列设置成了优先级队列，那么该队列它允许消息插队
//将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM内存的使用，如果未设置，队列将保留内存缓存以尽可能快地传递消息；
//有时候我们把这种队列叫：惰性队列
//arguments.put(“x-queue-mode”, “lazy”);
//设置队列版本。默认为版本1。
//版本1有一个基于日志的索引，它嵌入了小消息。
//版本2有一个不同的索引，可以在许多场景中提高内存使用率和性能，并为以前嵌入的消息提供了按队列存储。
//arguments.put(“x-queue-version”, 2);
// x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。
//arguments.put(“x-queue-master-locator”, QueueBuilder.LeaderLocator.clientLocal.getValue());
//-------------------------
//arguments.put(“x-expires”, 10000); //自动过期，10秒
//arguments.put(“x-message-ttl”, 10000); //自动过期，10秒，不会删除队列
//QueueBuilder 类里面有定义，设置队列溢出行为，当达到队列的最大长度时消息会发生什么，有效值是drop-head、reject-publish
//arguments.put(“x-max-length”, 5);
//arguments.put(“x-overflow”, QueueBuilder.Overflow.dropHead.getValue());
//表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，其他被忽略，false时消息循环分发给所有消费者(默认false)
//arguments.put(“x-single-active-consumer”, true);
// x-max-length-bytes，队列消息内容占用最大空间，受限于内存大小，超过该阈值则从队列头部开始删除消息；
//arguments.put(“x-max-length-bytes”, 10);
//参数是1到255之间的正整数，表示队列应该支持的最大优先级，数字越大代表优先级越高，没有设置priority优先级字段，那么priority字段值默认为0；如果优先级队列priority属性被设置为比x-max-priority大，那么priority的值被设置为x-max-priority的值。
//arguments.put(“x-max-priority”, 10);
//将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用;如果未设置，队列将保留内存缓存以尽可能快地传递消息；
//arguments.put(“x-queue-mode”, “lazy”);
arguments.put(“x-queue-version”, 2);
// x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。
arguments.put(“x-queue-master-locator”, QueueBuilder.LeaderLocator.clientLocal.getValue());
//---------------------------------------------
// Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments)
return new Queue(QUEUE, true, false, false, arguments);
}
@Bean
public Binding binding(DirectExchange directExchange, Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with(KEY);
}
}
|
| — |
实验durable 参数 重启rabbitmq-server，队列丢失
实验autodelete参数：加入接收者，发现停掉服务，那么久没有消费者了，对列就会自动删除
16. 消息可靠性投递
消息的可靠性投递就是要保证消息投递过程中每一个环节都要成功，那么这肯定会牺牲一些性能，性能与可靠性是无法兼得的；
如果业务实时一致性要求不是特别高的场景，可以牺牲一些可靠性来换取性能。
① 代表消息从生产者发送到Exchange;
② 代表消息从Exchange路由到Queue；
③ 代表消息在Queue中存储；
④ 代表消费者监听Queue并消费消息；
1、确保消息发送到RabbitMQ服务器的交换机上
可能因为网络或者Broker的问题导致①失败，而此时应该让生产者知道消息是否正确发送到了Broker的exchange中；
有两种解决方案：
第一种是开启Confirm（确认）模式；（异步）
第二种是开启Transaction（事务）模式；（性能低，实际项目中很少用）
2、确保消息路由到正确的队列
可能因为路由关键字错误，或者队列不存在，或者队列名称错误导致②失败。
使用return模式，可以实现消息无法路由的时候返回给生产者；
当然在实际生产环境下，我们不会出现这种问题，我们都会进行严格测试才会上线（很少有这种问题）；
另一种方式就是使用备份交换机（alternate-exchange），无法路由的消息会发送到这个备用交换机上；
3、确保消息在队列正确地存储
可能因为系统宕机、重启、关闭等等情况导致存储在队列的消息丢失，即③出现问题；
解决方案：
（1）、队列持久化
代码：
| QueueBuilder.durable(QUEUE).build(); |
|---|
（2）、交换机持久化
代码：
| ExchangeBuilder.directExchange(EXCHANGE).durable(true).build(); |
|---|
（3）、消息持久化
代码：
默认持久化
| MessageProperties messageProperties = new MessageProperties();
//设置消息持久化，当然它默认就是持久化，所以可以不用设置，可以查看源码
| messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT); |
|---|
（4）、集群，镜像队列，高可用
（5）确保消息从队列正确地投递到消费者
采用消息消费时的手动ack确认机制来保证；
如果消费者收到消息后未来得及处理即发生异常，或者处理过程中发生异常，会导致④失败。
为了保证消息从队列可靠地达到消费者，RabbitMQ提供了消息确认机制（message acknowledgement）；
#开启手动ack消息消费确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
消费者在订阅队列时，通过上面的配置，不自动确认，采用手动确认，RabbitMQ会等待消费者显式地回复确认信号后才从队列中删除消息；
如果消息消费失败，也可以调用basicReject()或者basicNack()来拒绝当前消息而不是确认。如果requeue参数设置为true，可以把这条消息重新存入队列，以便发给下一个消费者（当然，只有一个消费者的时候，这种方式可能会出现无限循环重复消费的情况，可以投递到新的队列中，或者只打印异常日志）；
17. 消息的幂等性
消息消费时的幂等性（消息不被重复消费）
同一个消息，第一次接收，正常处理业务，如果该消息第二次再接收，那就不能再处理业务，否则就处理重复了；
幂等性是：对于一个资源，不管你请求一次还是请求多次，对该资源本身造成的影响应该是相同的，不能因为重复的请求而对该资源重复造成影响；
以接口幂等性举例：
接口幂等性是指：一个接口用同样的参数反复调用，不会造成业务错误，那么这个接口就是具有幂等性的；
注册接口；
发送短信验证码接口；
比如同一个订单我支付两次，但是只会扣款一次，第二次支付不会扣款，这说明这个支付接口是具有幂等性的；
如何避免消息的重复消费问题？（消息消费时的幂等性）
全局唯一ID + Redis
生产者在发送消息时，为每条消息设置一个全局唯一的messageId，消费者拿到消息后，使用setnx命令，将messageId作为key放到redis中：setnx(messageId, 1)，若返回1，说明之前没有消费过，正常消费；若返回0，说明这条消息之前已消费过，抛弃；
具体代码参考以下代码；
参考代码：
| //1、把消息的唯一ID写入redis
boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(“idempotent:” + orders.getId(), String.valueOf(orders.getId())); //如果redis中该key不存在，那么就设置，存在就不设置
if (flag) { //key不存在返回true
//相当于是第一次消费该消息
//TODO 处理业务
System.out.println(“正常处理业务…” + orders.getId());
}
|
| — |