ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 其他分享> 文章详细

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

2021-03-01 09:59:02  阅读：249  来源： 互联网

标签：实战 rabbitTemplate 消费 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;
    }
}
二、消息持久化
  1. 在创建交换机和队列的时候，有个durable的参数，即是否持久化，如果设置为true，当rabbitmq服务器重启的时候，创建的交换机和队列均还存在着，不会丢失；
  2. 在发送消息的时候可以选择为该消息设置持久化，即消息体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)]

  • 代码目录如图所示，演示自动确认和手动确认：
    自动确认和手动确认.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实战指南》

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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有