不可能(？)NullPointerException-spring framework rabbit MQ，无法调用afterAckCallback

Question

我运行的Java应用程序使用RabbitMQ Server 3.8.9、spring-amqp-2.2.10.RELEASE和spring-rabbit-2.2.10.RELEASE。

我的测试用例执行以下操作：

< li >启动RabbitMQ服务器 < li >启动我的Java应用程序 < li >在我的Java应用程序上测试和验证一些功能 < li >正常停止我的Java应用程序 < li >正常停止RabbitMQ服务器 < li >多重复几次1-6

一切看起来都很好，除了有时在重新启动大约10分钟后，我在应用程序的日志中看到以下错误:

2021-02-05 12:52:46.498 UTC,ERROR,org.springframework.amqp.rabbit.connection.PublisherCallbackChannelImpl,null,rabbitConnectionFactory23,runWorker():1149,Failed to invoke afterAckCallback
java.lang.NullPointerException: null
    at org.springframework.amqp.rabbit.connection.PublisherCallbackChannelImpl.lambda$doHandleConfirm$1(PublisherCallbackChannelImpl.java:1027) ~[spring-rabbit.jar:2.2.10.RELEASE]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_181]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_181]
    at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_181]

进一步的分析没有指向任何具体的东西。RabbitMQ日志文件中没有错误，RabbitMQ服务器没有重新启动，在上面的时间戳期间，RabbitMQ日志中没有任何异常。

有问题的代码:

https://github . com/spring-projects/spring-amqp/blob/v 2 . 2 . 10 . release/spring-rabbit/src/main/Java/org/spring framework/amqp/rabbit/connection/publisher callbackchannelimpl . Java # l 1027

我的测试是自动化的，并作为CI管道的一部分运行。这个问题是间歇性的，我在沙箱中本地复制它时遇到了问题。

据我所知，我的Java应用程序的功能没有受到影响。

创建随处使用的RabbitMQ连接工厂的代码:

final CachingConnectionFactory connectionFactory = new CachingConnectionFactory(HOST_NAME);
connectionFactory.setChannelCacheSize(1);
connectionFactory.setPublisherConfirms(true);

这似乎是一个并发问题，但我不太确定如何找到它的根源。在大多数情况下，我们使用RabbitTemplate和其他Spring设施连接到RabbitMQ。

Spring的世界里有谁在兔子MQ护理方面有一些知识可以插话吗？

谢谢

Answer 1

您谈论的代码如下：

finally {
    try {
        if (this.afterAckCallback != null && getPendingConfirmsCount() == 0) {
                this.afterAckCallback.accept(this);
                this.afterAckCallback = null;
            }
        }
        catch (Exception e) {
            this.logger.error("Failed to invoke afterAckCallback", e);
        }
}

this.afterAckCallback属性周围确实可能存在竞争条件。我们可能会在一个线程中传递if（），但不同的线程会使this.afterAckCallback为null，因此我们在NPE中失败。我们必须将其值复制到本地变量，然后检查并执行接受（）。

随意提出针对Spring AMQP项目的GitHub问题：https://github.com/spring-projects/spring-amqp/issues

我们有一个争用条件，因为我们真的调用这个doHandleConfirm（），其异步逻辑来自进程MultipleAck（）中的循环。