MQ发布确认

简介: 生产者将信道设置成confirm模式,一旦信道进入confirm模式,所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始),一旦消息被投递到所有匹配的队列之后,broker就会发送一个确认给生产者(包含消息的唯一ID),这就使得生产者知道消息已经正确到达目的队列了,如果消息和队列是可持久化的,那么确认消息会在将消息写入磁盘之后发出,broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号,此外broker也可以设置basic.ack的multiple域,表示到这个序列号之前的所有消息都已经得到了处理。

发布确认原理



 生产者将信道设置成confirm模式,一旦信道进入confirm模式,所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始),一旦消息被投递到所有匹配的队列之后,broker就会发送一个确认给生产者(包含消息的唯一ID),这就使得生产者知道消息已经正确到达目的队列了,如果消息和队列是可持久化的,那么确认消息会在将消息写入磁盘之后发出,broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号,此外broker也可以设置basic.ack的multiple域,表示到这个序列号之前的所有消息都已经得到了处理。


 confirm模式最大的好处在于他是异步的,一旦发布一条消息,生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息,当消息最终得到确认之后,生产者应用便可以通过回调方法来处理该确认消息,如果RabbitMQ因为自身内部错误导致消息丢失,就会发送一条nack消息,生产者应用程序同样可以在回调方法中处理该nack消息


发布确认的策略  



开启发布确认的方法


发布确认默认是没有开启的,如果要开启需要调用方法confirmSelect,每当你要想使用发布确认,都需要在channel上调用该方法


cfbd55d1163c43868e522bbb42add35a.png


单个确认发布



这是一种简单的确认方式,它是一种同步确认发布的方式也就是发布一个消息之后只有它被确认发布,后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回,如果在指定时间范围内这个消息没有被确认那么它将抛出异常。


这种确认方式有一个最大的缺点就是:发布速度特别的慢,因为如果没有确认发布的消息就会阻塞所有后续消息的发布,这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

代码实现:工具类获取信道


public class untils {
    public static Channel getChannel() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.231.132");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}


实现:


 public static  void publishMessageIndivdually()throws Exception{
        String quequeName= UUID.randomUUID().toString();
        Channel channel = untils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for(int i=0;i<1000;i++)
        {
            String message=i+"";
            channel.basicPublish("",quequeName,null,message.getBytes(StandardCharsets.UTF_8));
            //服务端返回false或者超时时间内未返回,生产者可以消息重发
            boolean flag=channel.waitForConfirms();
//            if(flag)
//            {
//                System.out.println("消息发送成功");
//            }
        }
        long end=System.currentTimeMillis();
        System.out.println("发布"+1000+"个单独确认的消息,耗时"+(end-begin)+"ms");
    }


批量确认发布



上面那种方式非常慢,与单个等待确认消息相比,先发布一批消息然后一起确认可以极大地提高吞吐量,当然这种方式的缺点就是:当发生故障导致发布出现问题时,不知道是哪个消息出现问题了,我们必须将整个批处理保存在内存中,以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的,也一样阻塞消息的发布


public static  void publishMessageBatch()throws  Exception{
        String quequeName= UUID.randomUUID().toString();
        Channel channel = untils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //批量确认消息大小
        int batchSize = 100;
        // 未确认消息个数
        int outstandingMessageCount = 0;
        long begin=System.currentTimeMillis();
        for(int i=0;i<1000;i++)
        {
            String message=i+"";
            channel.basicPublish("",quequeName,null,message.getBytes(StandardCharsets.UTF_8));
            outstandingMessageCount++;
            if(outstandingMessageCount==batchSize)
            {
                channel.waitForConfirms();
                outstandingMessageCount=0;
            }
        }
        long end=System.currentTimeMillis();
        System.out.println("发布"+"1000个批量确认的消息,耗时"+(end-begin)+"ms");
    }



异步确认发布



异步确认虽然编程逻辑比上两个要复杂,但是性价比最高,无论是可靠性还是效率都没得说,他是利用回调函数来达到消息可靠性传递的,这个中间件也是通过函数回调来保证是否投递成功,下面就让我们来详细讲解异步确认是怎么实现的。



77d4dae105ad42089b440bda116f3fa7.png


如何处理异步未确认消息


最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列,比如说用ConcurrentLinkedQueue这个队列在confirm callbacks与发布线程之间进行消息的传递。

 

public static void publishMessaheAsync()throws  Exception{
        String quequeName= UUID.randomUUID().toString();
        Channel channel = untils.getChannel();
        channel.queueDeclare(quequeName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表,适合高并发的情况
         *1.轻松的将序号与消息进行关联
         * 2.轻松的批量的删除条目,只要给到序列号
         * 3.支持并发的访问
         */
        ConcurrentSkipListMap<Long,String> outstandingConfirms=new ConcurrentSkipListMap<>();
        /**
         * 确认到消息的一个回调
         * 1.消息的序列号
         * 2.true可以确认小于等于当前序列号的消息
         * false确认当前序列号的消息
         */
        ConfirmCallback ackCallback=(sequenceNumber,multiple)->{
            if(multiple)
            {
                //返回的是小于等于当前序列号未确认的消息,是一个map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                //消除该部分未确认的消息
                confirmed.clear();
            }
            else
            {
                //只清楚当前序列号的消息
                outstandingConfirms.remove(sequenceNumber);
            }
        };
        ConfirmCallback nackCallback=(sequenceNumber,multiple)->{
            String message=outstandingConfirms.get(sequenceNumber);
            System.out.println("发布消息"+message+"未被确认,序列号"+sequenceNumber);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback,nackCallback);
        long begin=System.currentTimeMillis();
        for(int i=0;i<1000;i++)
        {
            String message="消息"+i;
            /**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 全部都是未确认的消息体
             */
             outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
             channel.basicPublish("",quequeName,null,message.getBytes(StandardCharsets.UTF_8));
        }
        long end=System.currentTimeMillis();
        System.out.println("发布1000个异步确认的消息,耗时"+(end-begin)+"ms");
    }


结果:


9ffc122aeb8247b2a88ac681f1026009.png


单独发布消息

同步等待确认,简单,但吞吐量非常有限。

批量发布消息

批量同步等待确认,简单,合理的吞吐量,一旦出现问题但很难推断出是那条 消息出现了问题。

异步处理

最佳性能和资源使用,在出现错误的情况下可以很好地控制,但是实现起来稍微难些

相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
9天前
|
存储 关系型数据库 分布式数据库
PostgreSQL 18 发布,快来 PolarDB 尝鲜!
PostgreSQL 18 发布,PolarDB for PostgreSQL 全面兼容。新版本支持异步I/O、UUIDv7、虚拟生成列、逻辑复制增强及OAuth认证,显著提升性能与安全。PolarDB-PG 18 支持存算分离架构,融合海量弹性存储与极致计算性能,搭配丰富插件生态,为企业提供高效、稳定、灵活的云数据库解决方案,助力企业数字化转型如虎添翼!
|
8天前
|
存储 人工智能 Java
AI 超级智能体全栈项目阶段二:Prompt 优化技巧与学术分析 AI 应用开发实现上下文联系多轮对话
本文讲解 Prompt 基本概念与 10 个优化技巧,结合学术分析 AI 应用的需求分析、设计方案,介绍 Spring AI 中 ChatClient 及 Advisors 的使用。
366 130
AI 超级智能体全栈项目阶段二:Prompt 优化技巧与学术分析 AI 应用开发实现上下文联系多轮对话
|
8天前
|
人工智能 Java API
AI 超级智能体全栈项目阶段一:AI大模型概述、选型、项目初始化以及基于阿里云灵积模型 Qwen-Plus实现模型接入四种方式(SDK/HTTP/SpringAI/langchain4j)
本文介绍AI大模型的核心概念、分类及开发者学习路径,重点讲解如何选择与接入大模型。项目基于Spring Boot,使用阿里云灵积模型(Qwen-Plus),对比SDK、HTTP、Spring AI和LangChain4j四种接入方式,助力开发者高效构建AI应用。
356 122
AI 超级智能体全栈项目阶段一:AI大模型概述、选型、项目初始化以及基于阿里云灵积模型 Qwen-Plus实现模型接入四种方式(SDK/HTTP/SpringAI/langchain4j)
|
20天前
|
弹性计算 关系型数据库 微服务
基于 Docker 与 Kubernetes(K3s)的微服务:阿里云生产环境扩容实践
在微服务架构中,如何实现“稳定扩容”与“成本可控”是企业面临的核心挑战。本文结合 Python FastAPI 微服务实战,详解如何基于阿里云基础设施,利用 Docker 封装服务、K3s 实现容器编排,构建生产级微服务架构。内容涵盖容器构建、集群部署、自动扩缩容、可观测性等关键环节,适配阿里云资源特性与服务生态,助力企业打造低成本、高可靠、易扩展的微服务解决方案。
1342 8
|
2天前
|
存储 JSON 安全
加密和解密函数的具体实现代码
加密和解密函数的具体实现代码
190 136
|
7天前
|
监控 JavaScript Java
基于大模型技术的反欺诈知识问答系统
随着互联网与金融科技发展,网络欺诈频发,构建高效反欺诈平台成为迫切需求。本文基于Java、Vue.js、Spring Boot与MySQL技术,设计实现集欺诈识别、宣传教育、用户互动于一体的反欺诈系统,提升公众防范意识,助力企业合规与用户权益保护。
|
19天前
|
机器学习/深度学习 人工智能 前端开发
通义DeepResearch全面开源!同步分享可落地的高阶Agent构建方法论
通义研究团队开源发布通义 DeepResearch —— 首个在性能上可与 OpenAI DeepResearch 相媲美、并在多项权威基准测试中取得领先表现的全开源 Web Agent。
1440 87
|
7天前
|
JavaScript Java 大数据
基于JavaWeb的销售管理系统设计系统
本系统基于Java、MySQL、Spring Boot与Vue.js技术,构建高效、可扩展的销售管理平台,实现客户、订单、数据可视化等全流程自动化管理,提升企业运营效率与决策能力。