无法关闭jms侦听器，该监听器使用Runtime.exit、context.close、System.exit()向kafka spring引导应用程序发布消息

Question

我正在开发一个spring引导应用程序，它将使用

@JmsListener(id="abc", destination="${queueName}", containerFactory="defaultJmsListenerContainerFactory")

我有一个启动JmsListenerEndpointRegistry的listenerContainer。

On message将尝试用一些业务逻辑将相同的消息推送给kafka。海报代码是

 kafkaTemplate.send(kafkaProp.getTopic(), uniqueId, message)

现在，如果卡夫卡生产者失败，我希望我的引导应用程序被终止。所以我增加了一个自定义

setErrorHandler. 

所以我试过

    `System.exit(1)`, `configurableApplicationContextObject.close()`, `Runtime.getRuntime.exit(1)`.

但它们都不管用。下面是在System.exit(0)或其他日志之后生成的日志。

2018-05-24 12:12:47.981  INFO 18904 --- [       Thread-4] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@1d08376: startup date [Thu May 24 12:10:35 IST 2018]; root of context hierarchy
2018-05-24 12:12:48.027  INFO 18904 --- [       Thread-4] o.s.c.support.DefaultLifecycleProcessor  : Stopping beans in phase 2147483647
2018-05-24 12:12:48.028  INFO 18904 --- [       Thread-4] o.s.c.support.DefaultLifecycleProcessor  : Stopping beans in phase 0
2018-05-24 12:12:48.028  INFO 18904 --- [       Thread-4] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown
2018-05-24 12:12:48.028  INFO 18904 --- [       Thread-4] o.a.k.clients.producer.KafkaProducer     : Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
2018-05-24 12:12:48.044  INFO 18904 --- [       Thread-4] o.a.k.clients.producer.KafkaProducer     : Closing the Kafka producer with timeoutMillis = 30000 ms.

但是应用程序仍然在运行，下面是正在运行的线程。

Daemon Thread [Tomcat JDBC Pool Cleaner[14341596:1527144039908]] (Running)  
Thread [DefaultMessageListenerContainer-1] (Running)    
Thread [DestroyJavaVM] (Running)    
Daemon Thread [JMSCCThreadPoolMaster] (Running) 
Daemon Thread [RcvThread: com.ibm.mq.jmqi.remote.impl.RemoteTCPConnection@12474910[qmid=*******,fap=**,channel=****,ccsid=***,sharecnv=***,hbint=*****,peer=*******,localport=****,ssl=****]] (Running) 
Thread [Thread-4] (Running) 

我们非常感谢你的帮助。提前谢谢。我只想让应用程序退出。

下面是调用System.exit(1)之前的线程转储

"DefaultMessageListenerContainer-1" 
   java.lang.Thread.State: RUNNABLE
        at sun.management.ThreadImpl.getThreadInfo1(Native Method)
        at sun.management.ThreadImpl.getThreadInfo(ThreadImpl.java:174)
        at com.QueueErrorHandler.handleError(QueueErrorHandler.java:42)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:931)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:902)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:326)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:235)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1166)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1158)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1055)
        at java.lang.Thread.run(Thread.java:745)

Answer 1

两种解决方案让我解决了上面的问题。

解决方案1.在错误处理程序中获取所有线程，并将它们全部中断，然后存在系统。

ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 100);

for (ThreadInfo threadInfo : threadInfos) {
    Thread.currentThread().interrupt();
}
System.exit(1);

解决方案2.定义应用程序上下文管理器。喜欢

public class AppContextManager implements ApplicationContextAware {
    private static ApplicationContext _appCtx;

    @Override
    public void setApplicationContext(ApplicationContext ctx){
         _appCtx = ctx;
    }

    public static ApplicationContext getAppContext(){
        return _appCtx;
    } 

    public static void exit(Integer exitCode) {
        System.exit(SpringApplication.exit(_appCtx,() -> exitCode));
    }
}

然后使用相同的管理器退出错误处理程序。

Executors.newSingleThreadExecutor().execute(new Runnable() {

                public void run() {
                    jmsListenerEndpointRegistry.stop();
                    AppContextManager.exit(-1);
                }
            });

Answer 2

您应该使用一个线程转储来查看Thread [DefaultMessageListenerContainer-1] (Running)正在做什么。

如果卡夫卡的制片人失败了

什么样的失败？如果代理关闭，默认情况下，线程将在生产者库中阻塞多达60秒。

您可以通过设置max.block.ms生产者属性来缩短时间。