通过 Pulsar 源码彻底解决重复消费问题

通过,pulsar,源码,彻底解决,重复,消费,问题 · 浏览次数 : 218

小编点评

## Summary of the issue: The Pulsar consumer is experiencing repeated message consumption due to a configuration issue. **Key points:** * The problem occurs only when using the message listener and `ackTimeout` is set. * The `ackTimeout` is set to 30 seconds, which should theoretically prevent timeouts. * The consumer receives messages and acknowledges them within 2 seconds. * When the `ackTimeout` is commented out, the consumer does not experience any issues. * The issue appears to be related to the timing of the `ackTimeout` call and the callback execution within the message listener. **Possible solutions:** 1. Move the `ackTimeout` call to the callback function of the message listener. 2. Use the `nack` API instead of `ackTimeout` to achieve the same functionality. 3. Investigate the underlying cause of the timeouts and optimize the consumer configuration accordingly. **Additional notes:** * The issue appears to be related to a race condition between the `ackTimeout` call and the callback execution. * The bug seems to be specific to the message listener implementation. * The official Pulsar documentation does not provide clear guidance on handling `ackTimeout` and its implications.

正文

背景

最近真是和 Pulsar 杠上了,业务团队反馈说是线上有个应用消息重复消费。

而且在测试环境是可以稳定复现的,根据经验来看一般能稳定复现的都比较好解决。

定位问题

接着便是定位问题了,根据之前的经验让业务按照这几种情况先排查一下:

通过排查:1,2可以排除了。

  1. 没有相关日志
  2. 存在异常,但最外层也捕获了,所以不管有无异常都会 ACK。

第三个也在消费的入口和提交消息出计算了时间,最终发现都是在2s左右 ACK 的。

伪代码如下:

        Consumer consumer = client.newConsumer()
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .topic(topic)
                .ackTimeout(30, TimeUnit.SECONDS)
                .subscriptionName("my-sub")
                .messageListener(new MessageListener<byte[]>() {
                    @SneakyThrows
                    @Override
                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                        log.info("msg_id{}",msg.getMessageId().toString());
                        TimeUnit.SECONDS.sleep(2);
                        consumer.acknowledge(msg);
                    }
                })
                .subscribe();

那这就很奇怪了,因为代码里配置的 ackTimeout 是 30s,理论上来说是不会存在超时导致消息重发的。

为了排除是否是超时引起的,直接将业务代码注释掉了,等于是消息收到后立即就 ACK,经过测试发现这样确实就没有重复消费了。

为了再次确认是不是和 ackTimeout 有关,直接将 .ackTimeout(30, TimeUnit.SECONDS) 注释掉后测试,发现也没有重复消费了。

确认原因

既然如此那一定是和这个配置有关了,但看代码确实没有超时,为了定位具体原因只有去看 client 的源码了。

这里简单梳理下消息的消费的流程:

  1. 根据 .receiverQueueSize(1000) 的配置,默认情况下 broker 会直接给客户端推送 1000 条消息。
  2. 客户端将这 1000 条消息保存到内部队列中。
  3. 如果使用同步消费 receive() 时,本质上就是去 take 这个内部队列。
  4. 如果是使用的是 messageListener 异步消费并配置 ackTimeout,每当从队列里获得一条消息后便会把这条消息加入 UnAckedMessageTracker 内部的一个时间轮中,定时检测顶部是否存在消息,如果存在则会触发重新投递。
    4.1 加入时间轮后,异步调用我们自定义的事件,这个异步操作是提交到一个无界队列中由单个线程依次排队执行(这点是这次问题的关键)
  5. 业务 ACK 的时候会从时间轮中删除消息,所以如果消息 ACK 的足够快,在第四步就不会获取到消息进行重新投递。

整体流程如上图,代码细节如下图:

所以问题的根本原因就是写入时间轮(UnAckedMessageTracker)开始倒计时的线程和回调业务逻辑的不是同一个线程。

如果业务执行耗时,等到消息从那个单线程的无界队列中取出来的时候很有可能已经过了 ackTimeou 的时间,从而导致了超时重发。

也就是用户所理解的 ackTimeout 周期(应该进入回调时候开始计时)和 SDK 实现的不一致造成的。

之后我再次确认同样的代码换为同步消费是没有问题的,不会导致重复消费:

while (true) {
Message msg = consumer.receive();
            log.info(
                    "consumer Message received: " + new String(msg.getData()) + msg.getMessageId().toString());
            TimeUnit.SECONDS.sleep(2);
            consumer.acknowledge(msg);	
}

查看代码后发现同步代码的获取消息和加入 UnAckedMessageTracker 时间轮是同步的,也就不会出现超时的问题。

总结

所以其实 是messageListener 异步消费的 ackTimeout 的语义是有问题的,需要将加入 UnAckedMessageTracker 处移动到回调函数中同步调用。

我查看了最新的 2.11.x 版本的代码依然没有修复,正准备提个 PR 切换到 master 时才发现已经有相关的 PR 了,只是还没有发版。

修复的背景和思路也是类似的,具体参考:

https://github.com/apache/pulsar/pull/18911

其实业务中并不推荐使用 ackTimeout 这个配置了,不好预估时间从而导致超时,而且我相信大部分业务配置好 ackTImeout 后直到后续出问题的时候才想起来要改。
所以干脆一开始就不要使用。

在 go 版本的 SDK 中直接废弃掉了这个参数,推荐使用 nack API 替换。

与通过 Pulsar 源码彻底解决重复消费问题相似的内容:

通过 Pulsar 源码彻底解决重复消费问题

背景 最近真是和 Pulsar 杠上了,业务团队反馈说是线上有个应用消息重复消费。 而且在测试环境是可以稳定复现的,根据经验来看一般能稳定复现的都比较好解决。 定位问题 接着便是定位问题了,根据之前的经验让业务按照这几种情况先排查一下: 通过排查:1,2可以排除了。 没有相关日志 存在异常,但最外层

深入剖析:如何使用Pulsar和Arthas高效排查消息队列延迟问题

背景 前两天收到业务反馈有一个 topic 的分区消息堆积了: 根据之前的经验来看,要么是业务消费逻辑出现问题导致消费过慢,当然也有小概率是消息队列的 Bug(我们使用的是 pulsar)。 排查 通过排查,发现确实是在一点多的时候消息堆积了(后面是修复之后堆积开始下降)。 于是我在刚才堆积处查看了

Java智能之Spring AI:5分钟打造智能聊天模型的利器

通过本文的介绍,我们深入了解了Spring AI项目的优势和特性,以及在实际应用中的快速实战示例。Spring AI作为一个高度抽象化的人工智能应用程序开发框架,为开发者提供了便捷的模型支持、灵活的功能模块交换和优化能力。它不仅能将AI模型输出映射为POJO,还能与主流矢量数据库提供商无缝集成,从而...

Simple WPF: WPF自定义一个可以定义步长的SpinBox

通过WPF的按钮、文本输入框实现了一个简单的SpinBox数字输入用户组件并可以通过数据绑定数值和步长。本文中介绍了通过Xaml代码实现自定义组件的布局,依赖属性的定义和使用等知识点。

金仓数据库全攻略:简化部署,优化管理的全流程指南

通过本篇文章的学习和实践,我们深入了解了如何利用Docker技术快速部署KingbaseES数据库。从下载镜像到编写Docker Compose模板,再到容器的启动和管理,每一步都体现了现代化部署方式的便捷和高效。此外,我们还掌握了KSQL命令行工具的使用,这将极大地提升开发人员与数据库交互的效率。

SVG 标签的用法和应用场景

通过使用 标签,可以在 SVG 图像内部定义可重复使用的任意图案。这些图案可以通过 fill 属性或 stroke 属性进行引用。 使用场景 例如我们要在 中绘制大量的圆点点,可以通过重复使用 标签来实现。

5分钟带你了解RabbitMQ的(普通/镜像)集群

通过本文我们深入了解了RabbitMQ的集群模式及其优缺点。无论是普通集群还是镜像集群,都有其适用的场景和局限性。普通集群利用Erlang语言的集群能力,但消息可靠性和高可用性方面存在一定挑战;而镜像集群通过主动消息同步提高了消息的可靠性和高可用性,但可能会占用大量网络带宽。因此,在选择集群方案时,...

腾讯云 BI 数据分析与可视化的快速入门指南

通过本文的介绍,我们了解了腾讯云 BI 这款商业智能解决方案的基本功能和应用场景。从创建项目、连接数据源、数据表建模到页面搭建和推送功能的设置,我们通过一个互联网运营看板的案例,展示了如何快速入门并利用腾讯云 BI 进行数据分析和可视化。通过简单的数据编辑,我们可以轻松地设计报表,并实现数据的可视化...

赛博斗地主——使用大语言模型扮演Agent智能体玩牌类游戏。

通过大模型来实现多个智能体进行游戏对局这个想对已经比较成熟了无论是去年惊艳的斯坦福小镇还是比如metaGPT或者类似的框架都是使用智能体技术让大模型来操控,从而让大模型跳出自身“预测下一个token”的文字功能去探索更多的应用落地可能性。不过一直没有真正操作过,直到前段时间看到一个新闻《和GPT-4

基于腾讯元器搭建前端小助手

通过本文,我们了解了如何利用腾讯元器搭建一个前端助手智能体。通过使用插件和观察其使用效果,我们可以发现前端助手在解决问题和提供帮助方面的潜力。这个前端助手可以成为我们在前端开发过程中的得力助手,帮助我们提高工作效率和解决难题。随着智能技术的不断进步,我们可以期待前端助手在未来发展中的更多功能和应用。