Spring 集成：如何将 SingleConnectionFactory 与 ActiveMQ 一起使用？

Question

Spring Boot 2.3.1.RELEASE。

使用spring.jms.cache.enabled=true（默认），Spring 创建一个CachingConnectionFactory：

    @ConditionalOnProperty(prefix = "spring.jms.cache", name = "enabled", havingValue = "true",
            matchIfMissing = true)
    static class CachingConnectionFactoryConfiguration {

这很糟糕，因为it shouldn't be used with DefaultMessageListenerContainer。我认为这就是为什么我的某些消息会“丢失”直到它们突然重新出现的原因。

使用spring.jms.cache.enabled=false，Spring 创建一个ActiveMQConnectionFactory：

    @ConditionalOnProperty(prefix = "spring.jms.cache", name = "enabled", havingValue = "false")
    ActiveMQConnectionFactory jmsConnectionFactory(ActiveMQProperties properties,
            ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers) {
        return createJmsConnectionFactory(properties, factoryCustomizers);
    }

    private static ActiveMQConnectionFactory createJmsConnectionFactory(ActiveMQProperties properties,

这很糟糕，因为每次轮询都会创建一个到代理的新连接——数百个连接淹没了我的代理。

虽然我的问题的解决方案是使用SingleConnectionFactory。在AbstractPollingMessageListenerContainer.MessageListenerContainerResourceFactory我看到了：

    public Connection createConnection() throws JMSException {
        if (AbstractPollingMessageListenerContainer.this.sharedConnectionEnabled()) {
            Connection sharedCon = AbstractPollingMessageListenerContainer.this.getSharedConnection();
            return new SingleConnectionFactory(sharedCon).createConnection();
        }

所以我想我会：

Jms.channel(connectionFactory)
    .cacheLevel(DefaultMessageListenerContainer.CACHE_CONNECTION)

但事实证明，这个方法永远不会被调用，只有JmsAccessor.createConnection() 会创建一个ActiveMQConnectionFactory。我的缓存级别无效。

那么如何正确使用SingleConnectionFactory呢？

Answer 1

如果您有可变并发，缓存工厂只是 DMLC 的问题。

只需将SingleConnctionFactory 定义为@Bean 并使用Jms.channel(mySingleCF())...。

编辑

@SpringBootApplication
public class So63120705Application {

    public static void main(String[] args) {
        SpringApplication.run(So63120705Application.class, args).close(); // JVM should exit
    }

    @Bean
    public ApplicationRunner runner(ConnectionFactory jmsConnectionFactory, IntegrationFlowContext context,
            JmsTemplate template) {

        return args -> {
            SingleConnectionFactory connectionFactory = new SingleConnectionFactory(jmsConnectionFactory);

            IntegrationFlow flow = f -> f.channel(Jms.channel(connectionFactory)
                                .destination("foo"))
                        .handle(System.out::println);

            context.registration(flow)
                .id("jms")
                .addBean(connectionFactory)
                .register();

            template.convertAndSend("foo", "test");
            Thread.sleep(5_000);
        };
    }

}

spring.jms.cache.enabled=false