zoukankan      html  css  js  c++  java
  • rabbit

    本篇和大家分享的是关于rabbit的生产和消费方的一些实用的操作;正如文章标题,主要内容如producer的confirm和consumer的ack,这两者使用的模式都是用来保证数据完整性,防止数据丢失。

    • producer的confirm模式
    • consumer的ack模式

    producer的confirm模式

    首先,有这样一种业务场景1:a系统在做活动前,需要给用户的手机发送一条活动内容短信希望用户来参加,因为用户量有点大,所以通过往短信mq中插入数据方式,让短信服务来消费mq发短信;

    此时插入mq消息的服务为了保证给所有用户发消息,并且要在短时间内插入完成(因此用到了异步插入方式(快速)),我们就需要知道每次插入mq是否成功,如果不成功那我们可以收集失败的信息后补发(因此confirm模式排上了用场);如图设计:

     在springboot中可以使用基于amqp封装的工厂类来开启confirm模式,然后通过RabbitTemplate模板来设置回调函数,如下代码:

     1     ///region producer生产 - confirm模式
     2 
     3     public RabbitTemplate getRabbitTemplate(RabbitTemplate.ConfirmCallback confirmCallback) {
     4         return this.getRabbitTemplate(this.connectionFactory(), confirmCallback);
     5     }
     6 
     7     public RabbitTemplate getRabbitTemplate(CachingConnectionFactory connectionFactory, RabbitTemplate.ConfirmCallback confirmCallback) {
     8         RabbitTemplate template = new RabbitTemplate(connectionFactory);
     9         //product开启confirm模式
    10         connectionFactory.setPublisherConfirms(true);
    11         //设置confirm回调处理
    12         template.setConfirmCallback(confirmCallback);
    13         return template;
    14     }
    15     ///endregion

    这里通过RabbitTemplate.ConfirmCallback函数编程来传递我们自定义的回调方法,如下收集confirm返回的结果信息:

    1         RabbitUtil rabbitUtil = new RabbitUtil(this.getFirstNode().getLink());
    2         RabbitTemplate template = rabbitUtil.getRabbitTemplate((a, b, c) -> {
    3             System.out.println("firstNodeTpl - ConfirmCallback的Id:" + a.getId() + ";状态:" + b + ";信息:" + c);
    4         });

    最后再通过RabbitTemplate实例的convertAndSend方法发送mq信息,我们能够在日志中看到如下记录的信息:

    这里的状态true:表示send成功,false:表示send失败;通常false的时候信息c会有响应的错误提示,这里把网络断开,如下错误提示:

    consumer的ack模式

    再来,有这样一种场景2:短信服务去消费mq队列信息时,倘若服务调用的运营商发送短信接口异常了(短信运营商接口欠费),我们此时的短信是发送失败的,用户也收不到短信,但是在默认(默认开启ack)前提下mq消息已经被消费了rabbit中没有记录了(kafka例外);想要mq消息在业务逻辑异常时还存在,那么可以使用ack方式;

    在springboot中可以使用基于amqp封装的工厂类关闭自动ack模式,改为手动ack方式;只有当业务代码流程走完后,最后通过代码设置ack标识,来通知rabbit消息可以丢弃了;如果设置了手动模式后,又没有提交ack标识,那么mq中的消息一直存在无法释放(每次consumer消费后,rabbit会把noack的消息重复放入队列中):

     1     ///region consumer监听 - 手动ack
     2     public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
     3         return this.listenerContainerFactory(this.connectionFactory());
     4     }
     5 
     6     public SimpleRabbitListenerContainerFactory listenerContainerFactory(ConnectionFactory connectionFactory) {
     7         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
     8         factory.setConnectionFactory(connectionFactory);
     9         //代码手动ack
    10         factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    11         //开启消费者数量
    12         factory.setConcurrentConsumers(2);
    13         //每次接受数据量,默认250
    14         factory.setPrefetchCount(300);
    15         return factory;
    16     }
    17     ///endregion

    通过连接工厂设置手动ack方式,然后获取mq消息后,走完正常业务逻辑,最后再手动通知ack释放消息,如下:

    1     @RabbitListener(containerFactory = "firstNodeListener", queues = {"${shenniu.rabbits.firstNode.queue}"})
    2     private void firstNodeListener(String msg, Channel channel, Message message) {
    3         try {
    4             long deliverTag = message.getMessageProperties().getDeliveryTag();
    5             System.out.println("firstNodeListener - 消费消息 [" + deliverTag + "] - " + msg);
    6             channel.basicAck(deliverTag, true);
    7         } catch (Exception ex) {
    8         }
    9     }

    这里ack主要根据mq消息的唯一编号(deliverTag)来通知;如果我们不设置ack确认,那么消息状态会是这样如下rabbit管理后台:

  • 相关阅读:
    MT9v024总结
    常用芯片电路知识汇总
    octopress 如何添加youku视频和本地视频(octopress how to add a youku video or a local video)
    MT9M021/MT9M031总结
    TC358746AXBG/748XBG 桥接器说明
    mipi 调试经验
    1‘b0 什么意思
    MIPI总结和MIPI规格说明书
    octopress添加回到顶部按钮
    CentOS软件安装目录查找
  • 原文地址:https://www.cnblogs.com/wangrudong003/p/11436990.html
Copyright © 2011-2022 走看看