Spring amqp cachingConnectionFactory 一分钟未关闭

Question

我的客户端应用程序是 Spring-shell 应用程序，运行 SpringBoot 2.1.1.

我们正在按需创建 rabbitmq 连接 CachingConnectionFacotry

    CachingConnectionFactory connectionFactory = new CachingConnectionFactory(uri);

    connectionFactory.setPublisherReturns(true);

退出应用程序时，在@PreDestroy and/or onContextClosed 事件上调用 cachingConnectionFactory.destroy(); 不会关闭 cachingConnectionFactory 。应用程序退出需要一分钟。

下面的线程转储表明 cachingConnectionFactory 中有一个执行程序尚未关闭。

"spring-rabbit-deferred-pool-12" #29 prio=5 os_prio=31 tid=0x00007ff61245e000 nid=0x7503 waiting on condition [0x000070000fa57000]
   java.lang.Thread.State: TIMED_WAITING (parking)
 at sun.misc.Unsafe.park(Native Method)
 - parking to wait for  <0x000000076c928300> (a java.util.concurrent.SynchronousQueue$TransferStack)
 at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
 at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
 at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:362)
 at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:941)
 at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)

"spring-rabbit-deferred-pool-11" #28 prio=5 os_prio=31 tid=0x00007ff616461000 nid=0x7303 waiting on condition [0x000070000f954000]
   java.lang.Thread.State: TIMED_WAITING (parking)
 at sun.misc.Unsafe.park(Native Method)
 - parking to wait for  <0x000000076c928300> (a java.util.concurrent.SynchronousQueue$TransferStack)
 at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
 at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
 at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:362)
 at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:941)
 at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)

"process reaper" #14 daemon prio=10 os_prio=31 tid=0x00007ff614cd0800 nid=0x590b waiting on condition [0x000070000ec2d000]
   java.lang.Thread.State: TIMED_WAITING (parking)
 at sun.misc.Unsafe.park(Native Method)
 - parking to wait for  <0x00000006c078c1e0> (a java.util.concurrent.SynchronousQueue$TransferStack)
 at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
 at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
 at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:362)
 at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:941)
 at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)

我可以做些什么来立即退出应用程序吗？

Answer 1

我认为这与线程池无关...

public final void destroy() {
    super.destroy();
    resetConnection();
    if (getContextStopped()) {
        this.stopped = true;
        if (this.channelsExecutor != null) {
            this.channelsExecutor.shutdownNow();
        }
    }
}

shutDownNow() 在重置连接后调用，很明显，所有线程都是可中断的。

最好的猜测是某些东西（可能与网络有关）延迟了 resetConnection()。

Post 在遇到延迟时 complete 线程转储，可能在 GitHub 要点，或在 pastebin 或类似的地方。

Spring amqp cachingConnectionFactory 一分钟未关闭

Spring amqp cachingConnectionFactory not shutting down for a minute

spring-amqp