【初始RabbitMQ】发布订阅的实现

发布确认原理

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

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

简单的来说发布确认的作用就是防止数据丢失

发布确认的策略 

开启发布确认的方法

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

单个确认发布

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

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

public static void publicMessageIndividually() throws IOException, InterruptedException {Channel channel = RabbitMqUtils.getChannel();String queueName = UUID.randomUUID().toString();/**生成一个队列* 1.队列名称* 2.队列里面的信息是否持久化(磁盘)默认情况时在内存* 3.该队列是否只供一个消费者进行消费 是否消费共享 true是允许* 4.是否自动删除 最后一个消费者断开连接之后 该队列是否自动删除 true自动删除 false不自动删除* 5.其他参数 延迟消息等*/channel.queueDeclare(queueName,false,false,false,null);//开启发布确认channel.confirmSelect();long begin = System.currentTimeMillis();for (int i = 0; i < MESSAGE_COUNT; i++) {String message = i + "";channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));//服务器返回false或超时内未返回,生产者可以消息重新发送boolean flag = channel.waitForConfirms();if(flag){System.out.println("消息发布成功!");}}long end = System.currentTimeMillis();System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");}

批量确认发布

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

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

异步确认发布

异步确认发布通常指的是在消息发布过程中,消息的生产者(或发送方)在发送消息后,不需要等待消息被消费者(或接收方)确认接收,而是可以继续进行其他操作。这种异步确认的机制可以提高系统的并发性和效率

在具体实现上,异步确认发布通常涉及到以下几个步骤:

  1. 消息生产者将消息发送到消息队列或中间件中
  2. 消息队列或中间件在接收到消息后,会为每条消息分配一个唯一的序列号或ID
  3. 消息队列或中间件将消息投递到相应的消费者队列中
  4. 消费者从自己的消费者队列中拉取消息并进行处理
  5. 消费者在处理完消息后,会向消息队列或中间件发送一个确认消息,表示该消息已经被成功处理
  6. 消息队列或中间件在接收到消费者的确认消息后,会将该确认消息与原始消息的序列号或ID进行匹配,确认该消息已经被成功处理
  7. 消息队列或中间件会向消息生产者发送一个异步确认消息,通知消息生产者该消息已经被成功处理

 

public static void publishMessageAsync() throws IOException {Channel channel = RabbitMqUtils.getChannel();String queueName = UUID.randomUUID().toString();channel.queueDeclare(queueName,false,false,false,null);//开启发布确定channel.confirmSelect();/*** 线程安全有序的一个哈希表,适合用于高并发情况* 1、轻松将序号和消息进行关联* 2、轻松批量删除条目 只需要给到序列号* 3、支持高并发访问*/ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();//开始时间long begin = System.currentTimeMillis();ConfirmCallback ackCallback = (deliveryTag,multiple)->{//是否是批量处理if(multiple){//返回的是小于等于(如果是小于等于需要在参数中加true)当前序列号的未确认消息 是一个 mapConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag,true);confirmed.clear();}else{//只清除当前序列号的消息outstandingConfirms.remove(deliveryTag);}System.out.println("确认消息"+deliveryTag);};ConfirmCallback nackCallback = (deliveryTag,multiple)->{String message = outstandingConfirms.get(deliveryTag);System.out.println("发布的消息"+message+"未被确认,序列号"+deliveryTag);// System.out.println("未确认消息"+deliveryTag);};//准备消息的监听器 监听哪些消息成功 哪些消息失效channel.addConfirmListener(ackCallback,nackCallback);//批量发送消息for (int i = 0; i < MESSAGE_COUNT; i++) {String message = "消息"+i;/*** channel.getNextPublishSeqNo()获取下一个消息的序列号* 通过序列号与消息体进行一个关联* 全部都是未确认的消息体*/outstandingConfirms.put(channel.getNextPublishSeqNo(),message);channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));}long end = System.currentTimeMillis();System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");}

以上 3 种发布确认速度对比

单独发布消息:同步等待确认,简单,但吞吐量非常有限

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

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

花费时间对比

    public static void main(String[] args) throws IOException, InterruptedException {//单个确认发布ConfirmMessage.publicMessageIndividually();//发布1000个单独确认消息,耗时47298ms//批量确认发布ConfirmMessage.publishMessageBatch();//发布1000个批量确认消息,耗时866ms//异步确认发布ConfirmMessage.publishMessageAsync();//发布1000个批量确认消息,耗时104ms}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://xiahunao.cn/news/2809078.html

如若内容造成侵权/违法违规/事实不符,请联系瞎胡闹网进行投诉反馈,一经查实,立即删除!

相关文章

力扣随笔之移除元素(简单27)

思路&#xff1a;定义一个指针left&#xff0c;使该指针及该指针左边的数全部都不等于val&#xff0c;定义一个遍历指针i&#xff0c;若nums[i] val&#xff0c;则i自加&#xff0c;若nums[i] ! val&#xff0c;则将left&#xff0c;并将nums[i]的值赋给nums[left]&#xff0c…

手撕Transformer(三)| 基础Transformer整体结构代码解析,从宏观到微观

文章目录 1 理解重点2 背景介绍 假设3 过程及重要组件3.1 嵌入层和加入位置编码3.2 编码器 Encoder3.3.1 EncoderLayer编码层3.3.2 LayerNorm归一化层 3.3 解码器 Decoder3.4 整合连接Encoder和Decoder 4 完整可运行代码 1 理解重点 在之前一节我们已经介绍了Transformer的位置…

C语言--贪吃蛇

目录 1. 实现目标2. 需掌握的技术3. Win32 API介绍控制台程序控制台屏幕上的坐标COORDGetStdHandleGetConsoleCursorinfoCONSOLE_CURSOR_INFOSetConsoleCursorInfoSetConsoleCursorPositionGetAsyncKeyState 4. 贪吃蛇游戏设计与分析地图<locale.h>本地化类项setlocale函…

14. UE5 RPG使用GameplayTag

GameplayTag本来是应用在GAS游戏技能系统里面的&#xff0c;后来UE直接将其抽离出来&#xff0c;作为一个模块&#xff0c;现在可以不在GAS里也可以使用这个模块。比如&#xff0c;我需要判断一个射线拾取的物体&#xff0c;首先我需要判断这个actor是否存在&#xff0c;然后判…

Nest创建神经元,并显示电压变化曲线

nest 安装与介绍 NEST&#xff08;神经模拟工具&#xff09;最初是在 1990 年代后期开发的。它的主要目标是作为计算神经科学模拟器。它支持具有不同生物学细节水平的各种神经元和突触模型。例如&#xff0c;NEST 的神经元模型范围从泄漏积分和激发模型到详细的 Hodgkin-Huxle…

c++入门学习⑧——模板

目录 前言 基本介绍 什么是模板&#xff1f; 作用 特点 分类 函数模板 语法 使用方式 注意事项 函数模板和普通函数区别 普通函数和函数模板的调用规则 局限性 类模板 语法 类模板的成员函数创建时机 类模板实例化对象 类模板实例化对象做函数参数 类模板成…

优化测试稳定性的失败重试工具:pytest-rerunfailures详解!

一.前言 笔者在执行自动化测试用例时&#xff0c;会发现有时候用例失败并非代码问题&#xff0c;而是由于服务正在发版&#xff0c;导致请求失败&#xff0c;从而降低了自动化用例的稳定性&#xff0c;最后还要花时间定位到底是自身case的原因还是业务逻辑问题&#xff0c;还是…

Host ’‘ is not allowed to connect to this MySQL server

问题描述&#xff1a;Host ’‘ is not allowed to connect to this MySQL server 解决方案&#xff1a; 问题原因&#xff1a; 可能是你的帐号不允许从远程登陆&#xff0c;只能在localhost。这个时候只要在localhost的那台电脑&#xff0c;登入mysql后&#xff0c;更改 “my…

Java面试:Spring Cloud Alibaba

文章目录 引言I Spring Cloud Alibaba1.1 配置文件加载的优先级(由高到低)1.2 注册中心1.3 rpcII 高并发场景:缓存穿透/缓存失效/雪崩如何解决2.1 缓存穿透2.2 缓存击穿(失效)2.3 缓存雪崩引言 微服务涉及的中间件分布式事务事务的传播方式事务的隔离级别缓存穿透/缓存失效…

手把手教你Jenkins整合Jmeter实现自动化接口测试!

01、在机器上安装jmeter 下载&#xff1a;http://jmeter.apache.org/download_jmeter.cgi 这里我用了一台Windows安装jmeter用来写接口测试的脚本&#xff0c;启动前修改jmeter.properties 中 jmeter.save.saveservice.output_format值为xml。 编写接口测试脚本&#xff1a; 脚…

批量删除传参那些事

接口参数&#xff1a; public Object batchDeleteUsers(RequestBody List userIds) 工具提示传参&#xff1a; { “userIds”: [] } 错误&#xff01;&#xff01;&#xff01;讨逆猴子 报错&#xff1a;JSON parse error: Cannot deserialize value of type java.util.ArrayL…

信息抽取(UIE):使用自然语言处理技术提升证券投资决策效率

一、引言 在当今快速变化的证券市场中&#xff0c;信息的价值不言而喻。作为一名资深项目经理&#xff0c;我曾领导一个关键项目&#xff0c;旨在通过先进的信息抽取技术&#xff0c;从海量的文本数据中提取关键事件&#xff0c;如企业并购、新产品发布以及政策环境的变动。这些…

为什么会员模式是一种明智的扩张方式

会员模式看起来是一种有趣、令人兴奋且很酷的业务发展方式&#xff0c;但当您真正深入研究时&#xff0c;您可能会惊讶地发现它远不止于此。 会员资格为我们提供了一条道德扩展的途径。我们可以就地为客户提供服务。 这就是为什么会员模式可能成为您企业的下一步&#xff0c;…

微信小程序自制动态导航栏

写在前面 关于微信小程序导航栏的问题以及解决办法我已经在先前的文章中有提到&#xff0c;点击下面的链接即可跳转~ &#x1f90f;微信小程序自定义的导航栏&#x1f90f; 在这篇文章中我们需要做一个这样的导航栏&#xff01;先上效果图 &#x1f447;&#x1f447;&#x1f…

电商风控系统(flink+groovy+flume+kafka+redis+clickhouse+mysql)

一.项目概览 电商的防止薅羊毛的风控系统 需要使用 groovy 进行风控规则引擎的编写 然后其它技术进行各种数据的 存储及处理 薅羊毛大致流程 如果单纯使用 if else在业务代码中进行风控规则的编写 那么 维护起来会比较麻烦 并且跟业务系统强绑定不合适 所以一般独立成一个单…

数据结构(算法竞赛、蓝桥杯)--线段树+懒标记

1、B站视频链接&#xff1a;C02【模板】线段树懒标记 Luogu P3372 线段树 1_哔哩哔哩_bilibili 题目链接&#xff1a;P3372 【模板】线段树 1 - 洛谷 | 计算机科学教育新生态 (luogu.com.cn) void build(int p,int l,int r){tr[p]{l,r,w[l],0};if(lr)return;//叶子节点返回int…

【JVM】聊聊JVM生产环境常见的OOM问题

对于JVM来说&#xff0c;因为划分有固定的区域来执行字节码文件&#xff0c;无外乎&#xff0c;出问题的&#xff0c;也就是按照对应分分区会有常见的OOM问题。 栈 对于栈来说&#xff0c;栈的主要作用就是用于方法的执行&#xff0c;方法调用入栈、方法调出出栈。但是如果我…

【vue3语法】开发使用创建项目等

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、vue3创建vue3v2函数式、v3组合式api响应式方法ref、reactive计算属性conputed监听属性wacthvue3 选项式生命周期父子通信父传子defineProps编译宏 子传父de…

大学生考勤新趋势:技术驱动,数据说话

✍✍计算机编程指导师 ⭐⭐个人介绍&#xff1a;自己非常喜欢研究技术问题&#xff01;专业做Java、Python、微信小程序、安卓、大数据、爬虫、Golang、大屏等实战项目。 ⛽⛽实战项目&#xff1a;有源码或者技术上的问题欢迎在评论区一起讨论交流&#xff01; ⚡⚡ Java实战 |…

老隋蓝海项目temu跨境电商好不好做?

近年来&#xff0c;跨境电商成为我国对外贸易的新亮点&#xff0c;其中Temu作为拼多多旗下的新兴跨境电商平台&#xff0c;吸引了众多国内卖家参与。老隋作为行业内的知名人士&#xff0c;他对Temu跨境电商项目的评价备受关注。本文将分析老隋对Temu跨境电商的看法&#xff0c;…