首页 > 其他分享> > springboot rabbitmq高可用消息确认消费实战

springboot rabbitmq高可用消息确认消费实战

2021-03-01 09:59:02 作者：互联网

以下例子代码可在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（手动确认）。
无须确认流程图如下图所示，对于该模式，消息是否消费成功生产者端是不知道的，存在可能重复消费/消息消费失败的情况：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9lflJn1V-1614563540690)(https://upload-images.jianshu.io/upload_images/17109776-0291f9773eccf5c1.jpeg?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)]
代码目录如图所示，演示自动确认和手动确认：

对于设置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实战指南》

标签：实战,rabbitTemplate,消费,springboot,队列,确认,rabbitmq,发送,消息
来源： https://blog.csdn.net/Fiuty_Da/article/details/114253368