你知道吗？Pulsar 也会重复消费？

2023-02-28

代码 lock msg

背景许久没有分享Java相关的问题排查了，最近帮同事一起排查了一个问题：在使用Pulsar消费时，发生了同一条消息反复消费的情况。排查当他告诉我这个现象的时候我就持怀疑态度，根据之前使用的经验Pulsar在官方文档以及API中都解释过：只有当设置了消费的ackTimeout并超时消费时才会重复投递消

背景

许久没有分享 Java 相关的问题排查了，最近帮同事一起排查了一个问题：

在使用 Pulsar 消费时，发生了同一条消息反复消费的情况。

排查

当他告诉我这个现象的时候我就持怀疑态度，根据之前使用的经验 Pulsar 在官方文档以及 API 中都解释过：

只有当设置了消费的 ackTimeout 并超时消费时才会重复投递消息，默认情况下是关闭的，查看代码也确实没有开启。

那会不会是调用了 negativeAcknowledge() 方法呢(调用该方法也会触发重新投递)，因为我们使用了一个第三方库 https://github.com/majusko/pulsar-java-spring-boot-starter 只有当抛出异常时才会调用该方法。

查阅代码之后也没有地方抛出异常，甚至整个过程中都没看到异常产生;这就有点诡异了。

复现

为了捋清楚整个事情的来龙去脉，详细了解了他的使用流程;

其实也就是业务出现了 bug，他在消息消费时 debug 然后进行单步调试，当走完一次调试后，没多久马上又收到了同样的消息。

但奇怪的是也不是每次 debug 后都能重复消费，我们都说如果一个 bug 能 100% 完全复现，那基本上就解决一大半了。

所以我们排查的第一步就是完全复现这个问题。

为了排除掉是 IDEA 的问题(虽然极大概率不太可能)既然是 debug 的时候产生的问题，那其实转换到代码也就是 sleep 嘛，所以我们打算在消费逻辑里直接 sleep 一段时间看能否复现。

经过测试，sleep 几秒到几十秒都无法复现，最后索性 sleep 一分钟，神奇的事情发生了，每次都成功复现!

既然能成功复现那就好说了，因为我自己的业务代码也有使用到 Pulsar 的地方，为了方便调试就准备在自己的项目里再复现一次。

结果诡异的事情再次发生，我这里又不能复现了。

虽然这才是符合预期的，但这就没法调了呀。

本着相信现代科学的前提，我们俩唯一的区别就是项目不一样了，为此我对比了两边的代码。

  @PulsarConsumer(
            topic = xx,
            clazz = Xx.class,
            subscriptionType = SubscriptionType.Shared
    )
    public void consume(Data msg) {
        log.info("consume msg:{}", msg.getOrderId());
        Lock lock = redisLockRegistry.obtain(msg.getOrderId());
        if (lock.tryLock()) {
            try {
                orderService.do(msg.getOrderId());
            } catch (Exception e) {
                log.error("consumer msg:{} err:", msg.toString(), e);
            } finally {
                lock.unlock();
            }
        }

    }1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.