springboot之rabbit - producer的confirm和consumer的ack模式

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

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

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

producer的confirm模式

首先,有这样一种业务场景1:a系统在做活动前,需要给用户的手机发送一条活动内容短信希望用户来参加,因为用户量有点大,所以通过往短信mq中插入数据方式,让短信服务来消费mq发短信;
此时插入mq消息的服务为了保证给所有用户发消息,并且要在短时间内插入完成(因此用到了异步插入方式(快速)),我们就需要知道每次插入mq是否成功,如果不成功那我们可以收集失败的信息后补发(因此confirm模式排上了用场);如图设计:
image
在springboot中可以使用基于amqp封装的工厂类来开启confirm模式,然后通过RabbitTemplate模板来设置回调函数,如下代码:

    ///region producer生产 - confirm模式

    public RabbitTemplate getRabbitTemplate(RabbitTemplate.ConfirmCallback confirmCallback) {
        return this.getRabbitTemplate(this.connectionFactory(), confirmCallback);
    }

    public RabbitTemplate getRabbitTemplate(CachingConnectionFactory connectionFactory, RabbitTemplate.ConfirmCallback confirmCallback) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        //product开启confirm模式
        connectionFactory.setPublisherConfirms(true);
        //设置confirm回调处理
        template.setConfirmCallback(confirmCallback);
        return template;
    }
    ///endregion

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

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

最后再通过RabbitTemplate实例的convertAndSend方法发送mq信息,我们能够在日志中看到如下记录的信息:
image
这里的状态true:表示send成功,false:表示send失败;通常false的时候信息c会有响应的错误提示,这里把网络断开,如下错误提示:
image

consumer的ack模式

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

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

    ///region consumer监听 - 手动ack
    public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
        return this.listenerContainerFactory(this.connectionFactory());
    }

    public SimpleRabbitListenerContainerFactory listenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //代码手动ack
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //开启消费者数量
        factory.setConcurrentConsumers(2);
        //每次接受数据量,默认250
        factory.setPrefetchCount(300);
        return factory;
    }
    ///endregion

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

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

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

相关实践学习
容器服务Serverless版ACK Serverless 快速入门:在线魔方应用部署和监控
通过本实验,您将了解到容器服务Serverless版ACK Serverless 的基本产品能力,即可以实现快速部署一个在线魔方应用,并借助阿里云容器服务成熟的产品生态,实现在线应用的企业级监控,提升应用稳定性。
云原生实践公开课
课程大纲 开篇:如何学习并实践云原生技术 基础篇: 5 步上手 Kubernetes 进阶篇:生产环境下的 K8s 实践 相关的阿里云产品:容器服务 ACK 容器服务 Kubernetes 版(简称 ACK)提供高性能可伸缩的容器应用管理能力,支持企业级容器化应用的全生命周期管理。整合阿里云虚拟化、存储、网络和安全能力,打造云端最佳容器化应用运行环境。 了解产品详情: https://www.aliyun.com/product/kubernetes
目录
相关文章
|
4月前
|
Kubernetes Java 流计算
Flink application on k8s 有没有和 session 模式通过-C 指定额外的 jar 的相同功能啊?
Flink application on k8s 有没有和 session 模式通过-C 指定额外的 jar 的相同功能啊?
30 0
|
4月前
|
弹性计算 资源调度 Kubernetes
Flink三种集群模式,Standalone模式,Flink On YARN,Flink On K8S,这三种模式有啥优缺点,生产环境如何选择呢?
Flink三种集群模式,Standalone模式,Flink On YARN,Flink On K8S,这三种模式有啥优缺点,生产环境如何选择呢?
227 3
|
4月前
|
Java Maven
SpringBoot集成RabbitMQ-三种模式的实现
SpringBoot集成RabbitMQ-三种模式的实现
93 0
|
3月前
|
消息中间件 Java Spring
一文看懂Spring Boot整合Rabbit MQ实现多种模式的生产和消费
一文看懂Spring Boot整合Rabbit MQ实现多种模式的生产和消费
59 0
|
4月前
|
前端开发 JavaScript Java
SpringBoot+Vue(一)商品管理系统 模式介绍 、项目改造
SpringBoot+Vue(一)商品管理系统 模式介绍 、项目改造
68 0
|
4月前
|
消息中间件 SQL Java
spring boot Rabbit高级教程(三)
spring boot Rabbit高级教程
70 0
|
4月前
|
消息中间件 Java 测试技术
spring boot Rabbit高级教程(二)
spring boot Rabbit高级教程
26 0
|
4月前
|
消息中间件 存储 Java
spring boot Rabbit高级教程(一)
spring boot Rabbit高级教程
49 0
|
4月前
|
数据安全/隐私保护
ACK Acknowledgement 确认 AES Advanced Encryption Standard 高级加密标准 ATM Asynchronous Transfer Mode异步传输模式
ACK Acknowledgement 确认 AES Advanced Encryption Standard 高级加密标准 ATM Asynchronous Transfer Mode异步传输模式
23 0
|
6月前
|
Kubernetes 应用服务中间件 测试技术
kubernetes calico的两种网络模式BGP和IP-IP性能分析
kubernetes calico的两种网络模式BGP和IP-IP性能分析