迁移JMS事件的正确最终方法，监听Spring集成和Spring Boot

Question

我有一个JmsConfig配置类，它通过以下方式处理来自主题的JMS事件：

• 它定义了一个@Bean ConnectionFactory，其中包含一个ActiveMQ实现
• 它定义了一个@Bean JmsListenerContainerFactory实例化一个DefaultJmsListenerContainerFactory，并通过Boot的DefaultJmsListenerContainerFactoryConfigurer
• 它定义了一个@Bean MessageConverter，其中包含一个映射Jackson2MessageConverter，并设置一个自定义对象映射器
• 我在我的服务方法上使用了指向myfactory的@JmsListener注释。这是我对这个主题的唯一用途，仅限订阅

现在我想转到Spring集成。在阅读了大量内容之后，如果我不需要双向使用（丢弃网关）或轮询机制（丢弃@InboundChannelAdapter），我将使用传统的XML配置措辞，选择一个消息驱动的通道适配器。我发现Java习语应该通过新的Spring Integration DSL库来实现，因此，我寻找合适的代码片段。

似乎JmsMessageDrivenChannelAdapter是正确的等价物，我找到了一种方法：

IntegrationFlows。来自（Jms.messageDriverChannelAdapter（…））

但问题是，它只接受ActiveMQ连接工厂或AbstractMessageListenerContainer，而不接受my boot预配置的JmsListenerContainerFactory！

如何最终实现这一点？




Answer 1

甚至还有一个更好的方法。我很惊讶Gary没有评论它。有一个开箱即用的构建器叫做Jms.container（...）。

@Bean
public IntegrationFlow jmsMyServiceMsgInboundFlow(ConnectionFactory connectionFactory, MessageConverter jmsMessageConverter, MyService myService, JmsProperties jmsProperties, @Value("${mycompany.jms.destination.my-topic}") String topicDestination){

      JmsProperties.Listener jmsInProps = jmsProperties.getListener();

      return IntegrationFlows.from(
                                Jms.messageDrivenChannelAdapter(  Jms.container(connectionFactory, topicDestination)
                                                                     .pubSubDomain(false)
                                                                     .sessionAcknowledgeMode(jmsInProps .getAcknowledgeMode().getMode())
                                                                     .maxMessagesPerTask(1)
                                                                     .errorHandler(e -> e.printStackTrace())
                                                                     .cacheLevel(0)
                                                                     .concurrency(jmsInProps.formatConcurrency())
                                                                     .taskExecutor(Executors.newCachedThreadPool())
                                                                     .get()))
                                   )
                                   .extractPayload(true)
                                   .jmsMessageConverter(jmsMessageConverter)
                                   .destination(topicDestination)
                                   .autoStartup(true)
                                    //.errorChannel("NOPE")
                             )
                             .log(LoggingHandler.Level.DEBUG)
                             .log()
                             .handle(myService, "myMethod", e -> e.async(true).advice(retryAdvice()))
                             .get();

Answer 2

JmsListenerContainerFactory是特定于@JmsListener的，它是用于配置DefaultMessageListenerContainer的更高级别抽象。Boot不为原始DefaultMessageListenerContainer提供自动配置选项；你得自己接线。但你仍然可以使用启动属性。。。

@Bean
public IntegrationFlow flow(ConnectionFactory connectionFactory, 
                            JmsProperties properties) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(container(connectionFactory, properties)))
            ...
            .get();
}

private DefaultMessageListenerContainer container(ConnectionFactory connectionFactory, 
                                                  JmsProperties properties) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConcurrentConsumers(properties.getListener().getConcurrency());
    container.setMaxConcurrentConsumers(properties.getListener().getMaxConcurrency());
    ...
    return container;
}