java - Tomcat 在使用 Spring Integration Java DSL 时挂起关闭

Question

我遇到一个问题，尝试正常关闭 Tomcat (8) 永远不会完成，因为 DefaultMessageListenerContainer 似乎被无限期地阻止(或循环)。

我一直在谷歌上搜索解决方案，但我发现的任何类似方法都没有用。这包括(但不限于):

• 使用configureListenerContainer()设置容器的taskExecutor
• 使用 Messages.queue() 而不是 Messages.direct()
• 将 ActiveMQConnectionFactory 包装在 CachingConnectionFactory 中

一个简单的 Servlet 3.0 示例:

compile 'org.springframework.integration:spring-integration-core:4.3.6.RELEASE'
compile 'org.springframework.integration:spring-integration-jms:4.3.6.RELEASE'
compile 'org.springframework.integration:spring-integration-java-dsl:1.2.1.RELEASE'

初始化器:

public class ExampleWebApp implements WebApplicationInitializer {
    @Override
    public void onStartup(final ServletContext servletContext) throws ServletException {
        final AnnotationConfigWebApplicationContext springContext = new AnnotationConfigWebApplicationContext();
        springContext.register(ExampleConfig.class);

        servletContext.addListener(new ContextLoaderListener(springContext));

        final ServletRegistration.Dynamic registration = servletContext.addServlet("example", new HttpRequestHandlerServlet());
        registration.setLoadOnStartup(1);
        registration.addMapping("/status");
    }
}

配置:

@Configuration
@EnableIntegration
public class ExampleConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        final ActiveMQConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory();
        mqConnectionFactory.setBrokerURL("tcp://host:port");
        mqConnectionFactory.setUserName("----");
        mqConnectionFactory.setPassword("----");

        return mqConnectionFactory;
    }

    @Bean
    public Queue testQueue() {
        return new ActiveMQQueue("test.queue");
    }

    @Bean
    public MessageChannel testReceiveChannel() {
        return MessageChannels.direct().get();
    }

    @Bean
    public IntegrationFlow pushMessageInboundFlow() {
        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(connectionFactory())
                         .destination(testQueue()))
                .log()
                .transform(new JsonToObjectTransformer(TestMessageObject.class))
                .channel(testReceiveChannel())
                .get();
    }

    /** Example message object */
    public static class TestMessageObject {
        private String text;

        public String getText() {
            return text;
        }

        public void setText(final String text) {
            this.text = text;
        }
    }
}

如果我尝试通过 catalina.sh 脚本停止它(例如，在 Intellij 中按“停止”)，它永远不会结束存在。到目前为止，我能够完成关闭的唯一方法是通过“手动”在关机时销毁 JmsMessageAdapters，通过一个小助手类:

public class JmsMessageListenerContainerLifecycleManager {
    private static final Logger LOG = LoggerFactory.getLogger(JmsMessageListenerContainerLifecycleManager.class);

    @Autowired
    private List<IntegrationFlow> mIntegrationFlows;

    @PreDestroy
    public void shutdownJmsAdapters() throws Exception {
        LOG.info("Checking {} integration flows for JMS message adapters", mIntegrationFlows.size());

        for (IntegrationFlow flow : mIntegrationFlows) {
            if (flow instanceof StandardIntegrationFlow) {
                final StandardIntegrationFlow standardFlow = (StandardIntegrationFlow) flow;

                for (Object component : standardFlow.getIntegrationComponents()) {
                    if (component instanceof JmsMessageDrivenChannelAdapter) {
                        final JmsMessageDrivenChannelAdapter adapter = (JmsMessageDrivenChannelAdapter) component;

                        LOG.info("Destroying JMS adapter {}", adapter.getComponentName());
                        adapter.destroy();
                    }
                }
            }
        }
    }
}

虽然这行得通，但感觉绝对是错误的解决方案。

之前我是用spring-integration的xml配置，没有这个问题。我错过了什么？

Answer 1

呃!这绝对是一个错误。看起来您可以正确解决它。

尽管考虑销毁那里的任何 DisposableBean。

我正在将修复添加到 Spring Integration Java DSL。我们将在 Spring Integration 4.3.9 之后发布下一个 1.2.2。

Spring Integration 5.0 将在明天的 M3 版本中修复。