- 以下例子代码可在github或者在gitee下载
github：代码链接
gitee：代码链接 - 前两篇博文：
springboot rabbitmq入门使用
springboot rabbitmq不同交换机类型实战
RabbitMQ的高可用主要体现在消息的发送、传输和接收的过程中，可以保证消息成功发送、不会丢失，以及被确认消费/不重复消费。
- 对于消息是否发送成功，主要是针对生产者端的消息生产确认机制；
- 对于消息不会丢失，主要是rabbitmq消息持久化机制；
- 对于消息确认消费/不重复消费，主要是针对消费者端对消息的确认消费机制。
一、消息生产确认机制
对于消息是否发送成功，在rabbitmq自定义操作组件中可以统一设置消息生产确认相关逻辑rabbitTemplate.setConfirmCallback和rabbitTemplate.setReturnCallback。
@Slf4j
@Configuration
public class RabbitmqConfig {
//自定义配置RabbitMQ发送消息的操作组件RabbitTemplate
@Bean
public RabbitTemplate rabbitTemplate(){
//设置“发送消息后进行确认”
connectionFactory.setPublisherConfirms(true);
//设置“发送消息后返回确认信息”
connectionFactory.setPublisherReturns(true);
//构造发送消息组件实例对象
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMandatory(true);
//发送消息后，如果发送成功，则输出“消息发送成功”的反馈信息
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData,ack,cause));
//发送消息后，如果发送失败，则输出“消息发送失败-消息丢失”的反馈信息
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message));
//定义消息传输的格式为JSON字符串格式
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
//最终返回RabbitMQ的操作组件实例RabbitTemplate
return rabbitTemplate;
}
}
二、消息持久化
- 在创建交换机和队列的时候，有个durable的参数，即是否持久化，如果设置为true，当rabbitmq服务器重启的时候，创建的交换机和队列均还存在着，不会丢失；
- 在发送消息的时候可以选择为该消息设置持久化，即消息体Message的deliveryMode设置为MessageDeliveryMode.PERSISTENT持久化，当消息来不及消费rabbitmq服务器重启，那么消息依旧存在，如果将所有消息都设置持久化，那么会影响性能，内存和磁盘的读写速度差异很大。
三、消息确认消费机制
如何保证消息能够被准备消费、不重复消费，RabbitMQ提供了消息确认机制，即ACK模式。RabbitMQ的消息确认机制有3种，分别是NONE（无须确认）、AUTO（自动确认）和MANUAL（手动确认）。
-
无须确认流程图如下图所示，对于该模式，消息是否消费成功生产者端是不知道的，存在可能重复消费/消息消费失败的情况：
无需确认.jpeg -
代码目录如图所示，演示自动确认和手动确认：
自动确认和手动确认.png
对于设置ACK模式，可以在yaml配置文件中设置spring.rabbitmq.listener.simple.acknowledge-mode: xxx，也可以在声明的监听器Bean中设置，用简单监听器SimpleRabbitListenerContainerFactory即可：
@Slf4j
@Configuration
public class RabbitmqConfig {
/**
* 确认消费模式为自动确认机制-AUTO,采用直连传输directExchange消息模型
*/
@Bean
public SimpleRabbitListenerContainerFactory singleListenerContainerAuto(){
//定义消息监听器所在的容器工厂
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//设置容器工厂所用的实例
factory.setConnectionFactory(connectionFactory);
//设置消息在传输中的格式，在这里采用JSON的格式进行传输
factory.setMessageConverter(new Jackson2JsonMessageConverter());
//设置并发消费者实例的初始数量。在这里为1个
factory.setConcurrentConsumers(1);
//设置并发消费者实例的最大数量。在这里为1个
factory.setMaxConcurrentConsumers(1);
//设置并发消费者实例中每个实例拉取的消息数量-在这里为1个
factory.setPrefetchCount(1);
//确认消费模式为自动确认机制
factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
return factory;
}
/**
* 确认消费模式为手动确认机制-MANUAL,采用直连传输directExchange消息模型
*/
@Bean
public SimpleRabbitListenerContainerFactory singleListenerContainerManual(){
//定义消息监听器所在的容器工厂
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//设置容器工厂所用的实例
factory.setConnectionFactory(connectionFactory);
//设置消息在传输中的格式，在这里采用JSON的格式进行传输
factory.setMessageConverter(new Jackson2JsonMessageConverter());
//设置并发消费者实例的初始数量。在这里为1个
factory.setConcurrentConsumers(1);
//设置并发消费者实例的最大数量。在这里为1个
factory.setMaxConcurrentConsumers(1);
//设置并发消费者实例中每个实例拉取的消息数量-在这里为1个
factory.setPrefetchCount(1);
//确认消费模式为自动确认机制
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
return factory;
}
}
（1）自动确认模式
自动确认模式流程图如图所示，RabbitMQ内置组件通知生产者端，当消息成功消费/消费失败都会通知：
auto确认.jpeg
对于自动确认模式，在消费者端可以看到和普通的消息队列没什么区别，而手工确认消费模式则比较灵活。
- 确认消费模式为自动确认机制-AUTO,采用直连传输directExchange消息模型-生产者
@Slf4j
@Component
public class AutoAckPublisher {
//定义RabbitMQ消息操作组件RabbitTemplate
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发送消息
*/
public void sendMsg(Order order) {
try {
//设置交换机
rabbitTemplate.setExchange(RabbitMqConstants.AUTO_ACKNOWLEDGE_EXCHANGE);
//设置路由
rabbitTemplate.setRoutingKey(RabbitMqConstants.AUTO_ACKNOWLEDGE_ROUTING_KEY);
//发送消息
rabbitTemplate.convertAndSend(order);
log.info("确认消费模式为自动确认机制-消息模型DirectExchange-one-生产者-发送消息：{} ",order);
}catch (Exception e){
log.error("确认消费模式为自动确认机制-消息模型DirectExchange-one-生产者-发送消息:{},发生异常：{} ",order, e);
}
}
}
- 确认消费模式为自动确认机制-AUTO,采用直连传输directExchange消息模型-消费者
@Slf4j
@Component
public class AutoAckConsumer {
@RabbitListener(queues = RabbitMqConstants.AUTO_ACKNOWLEDGE_QUEUE, containerFactory = "singleListenerContainerAuto")
public void consumeMsg(Order order) {
try {
log.info("基于AUTO的自动确认消费模式-消费者监听消费消息-内容为：{} ",order);
}catch (Exception e){
log.error("基于AUTO的自动确认消费模式-消费者监听消费消息:{},发生异常：", order, e);
}
}
}
（2）手工确认流程图如图所示，当消息处理过程中出现异常的时候，需要手工确认处理该异常消息，该消息是否重新归入队列等处理。
manual确认.jpeg
- 确认消费模式为手动确认机制-MANUAL,采用直连传输directExchange消息模型-生产者
@Slf4j
@Component
public class ManualAckPublisher {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发送消息
*/
public void sendMsg(Order order) {
try {
rabbitTemplate.setExchange(RabbitMqConstants.MANUAL_ACKNOWLEDGE_EXCHANGE);
rabbitTemplate.setRoutingKey(RabbitMqConstants.MANUAL_ACKNOWLEDGE_ROUTING_KEY);
rabbitTemplate.convertAndSend(order);
log.info("确认消费模式为手动确认机制-消息模型DirectExchange-one-生产者-发送消息：{} ", order);
}catch (Exception e){
log.error("确认消费模式为手动确认机制-消息模型DirectExchange-one-生产者-发送消息:{},发生异常：{} ", order, e);
}
}
}
- 确认消费模式为手动确认机制-MANUAL,采用直连传输directExchange消息模型-消费者
在监听到消息并且消息成功处理完之后，通过basicAck来确认消息成功消费，当捕获到异常的时候即该消息处理失败的时候，有两种方式，一种是拒绝该消息并且消息重新归入队列中，另一种是拒绝该消息并且丢弃掉，一般情况下重新归入队列，还是会出现异常没法消费掉，除非把异常修复了才行，并且在未修复该异常的情况下，后面的消息会被堵塞住没办法消费，将消息重新归入队列中或许不是一个好的选择。
一般情况下可以保留该消息的信息然后把消息丢弃掉，最后重新发送消息；或者把该消息丢入到死信队列中，不对该死信队列进行监听，最后在rabbitmq管理后台取出该消息/重新监听该消息重新发送到原先队列进行消费，修复好异常情况再发送消息进行处理，保证消息成功消费。
@Slf4j
@Component
public class ManualAckConsumer {
@RabbitListener(queues = RabbitMqConstants.MANUAL_ACKNOWLEDGE_QUEUE, containerFactory = "singleListenerContainerManual")
public void consumeMsg(Order order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) Long tag) throws IOException {
try {
log.info("基于MANUAL的手工确认消费模式-消费者监听消费消息,消息投递标记：{},内容为：{} ", tag, order);
//抛异常,归入使得消息重新归入队列
//int num = 1 / 0;
//执行完业务逻辑后，手动进行确认消费，其中第一个参数为：消息的分发标识(全局唯一);第二个参数：是否允许批量确认消费
channel.basicAck(tag, false);
}catch (Exception e){
//第二个参数reueue重新归入队列,true的话会重新归入队列,需要人为地处理此次异常消息,重新归入队列也会继续异常
channel.basicReject(tag, true);
log.error("基于MANUAL的手工确认消费模式-消费者监听消费消息:{},消息投递标签：{},发生异常：", order, tag, e);
}
}
}
出现异常重新归入队列的情况，如图所示，显示有unacked 1条消息，下面有get messages，当点击的时候发现提示queue is empty队列为空，确实准备消费的消息为0条，正在消费的消息一直是unacked状态无法取出。
unacked消息.png
取不出来.png
这个时候只能停止监听重启项目，这个在线上不是好的办法，停止监听之后消息变为ready状态，这个时候可以取出，可以看到提示“取出消息是毁灭性的操作”。
ready状态消息.png
取出消息.png
四种取出消息的模式，分别为：不确认消息重新归入队列、确认消息不重新归入队列、拒绝该消息重新归入队列、拒绝该消息不重新归入队列。当取出消息可以看到消息的内容。
取出消息模式.png
消息内容.png
对于确认消息消费，避免消息异常出现上述情况，可以用死信队列来处理，捕获异常消息，发送消息到死信队列，不监听该队列的消息，最后修复异常重新发送消息到原先队列进行消费，详情请看下篇博文
参考资料：
《分布式中间件实战》
《rabbitmq实战指南》
