为什么 JmsListeners 不能共享一个公共的 Jms 连接？

Question

我以这种方式创建消息处理程序：

@JmsListeners(
  JmsListener(destination = "queue1"),
  JmsListener(destination = "queue2"),
  JmsListener(destination = "queue3"),
  JmsListener(destination = "queue4")
)
fun handleMessage(message: String) {
  // handle a message
}

当我检查我的消息代理时，我看到我的应用已建立 4 个连接。 不幸的是，我对来自 MQ 管理员的连接数量有限制，所以我希望消息处理程序只使用 1 个连接。

在检查了 Spring Jms 内部之后，我发现 DefaultMessageListenerContainer has an ability to use a shared connection。但问题是Spring的DefaultMessageListenerContainerFactory为每个@JmsListener创建了一个单独的DefaultMessageListenerContainer。

同时，JMS API 允许从单个JMSContext 创建多个JMSConsumers，例如

val jmsContext = connectionFactory.createContext(Session.SESSION_TRANSACTED)
val consumer1 = jmsContext.createConsumer(jmsContext.createQueue("queue1"))
val consumer2 = jmsContext.createConsumer(jmsContext.createQueue("queue2"))

如何设置JmsListeners 以共享公共连接？如果这不可能，Spring 有合理的理由吗？

Answer 1

我想将此作为评论发布，但我还需要显示代码，作为妥协...

我想你已经这样做了。这就像剥洋葱一样。

看起来好像不能通过application.properties-https://docs.spring.io/spring-boot/docs/2.4.1/reference/htmlsingle/#common-application-properties-integration控制sharedConnection

也不可以直接在代码中设置，但可能有办法影响它。使用以下一种或多种机制。

对于侦听器，您必须指定容器工厂。顺便提一句。我不喜欢硬编码的字符串，所以在application.properties 中有一个my.queue.name1 属性。

import org.springframework.jms.annotation.JmsListener;
    ...

    // Set upper concurrency limit on listener
    @JmsListener(destination = "${my.queue.name1}", containerFactory = "myListenerFactory", concurrency = "2")
    ...

回顾一下，您将需要一个用于自定义工厂的 bean，您可以在其中控制如何创建连接。 @Override createContainerInstance 方法是可能的，但是 DefaultMessageListenerContainer::sharedConnectionEnabled 方法是受保护的，并且是最终的，因此无法调用或覆盖。所以你需要使用其他方式来限制可用的连接数...

import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
...    

    @Bean("myListenerFactory")
    public DefaultJmsListenerContainerFactory myCustomisedListenerFactory() {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory() {
            @Override
            protected DefaultMessageListenerContainer createContainerInstance() {
                DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();

                // This may the way to control how many concurrent consumers are created
                dmlc.setMaxConcurrentConsumers(2);
                return dmlc;
            }
        };
        factory.setConnectionFactory(connectionFactory);

        // Set the concurrency lower limit to upper limit
        factory.setConcurrency("1-2");

        // Plus any other customisations
        ...

        return factory;
    }

还有你的自动有线连接工厂。

import import javax.jms.ConnectionFactory;
...

    @Autowired
    private ConnectionFactory connectionFactory;