使用spring集成确保jms消费者关闭的正确方法是啥？

Posted 2023-03-10

【中文标题】使用spring集成确保jms消费者关闭的正确方法是啥？

【英文标题】：What's the right way to ensure jms consumers are closed using spring integration?使用spring集成确保jms消费者关闭的正确方法是什么？

【发布时间】：2012-05-12 16:05:06

【问题描述】：

我正在使用 spring 集成来调用活动 mq 另一端的服务。我的配置如下：

<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg>
            <bean class="org.apache.activemq.ActiveMQConnectionFactory"
                  p:brokerURL="$risk.approval.queue.broker"
                  p:userName="$risk.approval.queue.username"
                  p:password="$risk.approval.queue.password"
                    />
        </constructor-arg>
        <property name="reconnectOnException" value="true"/>
        <property name="sessionCacheSize" value="100"/>
    </bean>

    <!-- create and close a connection to prepopulate the pool -->
    <bean factory-bean="jmsConnectionFactory" factory-method="createConnection" class="javax.jms.Connection"
          init-method="close" />

    <integration:channel id="riskApprovalRequestChannel"/>
    <integration:channel id="riskApprovalResponseChannel"/>

    <jms:outbound-gateway id="riskApprovalServiceGateway"
                          request-destination-name="$risk.approval.queue.request"
                          reply-destination-name="$risk.approval.queue.response"
                          request-channel="riskApprovalRequestChannel"
                          reply-channel="riskApprovalResponseChannel"
                          connection-factory="jmsConnectionFactory"
                          receive-timeout="5000"/>

    <integration:gateway id="riskApprovalService" service-interface="com.my.super.ServiceInterface"
                         default-request-channel="riskApprovalRequestChannel"
                         default-reply-channel="riskApprovalResponseChannel"/>

我注意到，使用此配置，消费者创建来从活动 mq 获取匹配请求永远不会关闭。每个请求都会增加消费者计数。

我可以通过添加来阻止这种情况发生

<property name="cacheConsumers" value="false" />

到 CachingConnectionFactory。

但是根据 CachingConnectionFactory 的 java 文档：

请注意，持久订阅者只会被缓存，直到合乎逻辑 会话句柄的关闭。

这表明会话永远不会关闭。

这是一件坏事吗？有没有更好的方法来阻止消费者堆积？

干杯， 彼得

【参考方案1】：

首先，您不需要工厂 bean 上的 init 方法 - 它什么都不做 - 会话工厂只有一个连接，并且在其上调用 close() 是无操作的。 （CCF 是 SingleConnectionFactory 的子类）。

第二；缓存消费者是默认设置；会话永远不会关闭，除非会话数超过 sessionCacheSize（您已设置为 100）。

当在缓存会话上调用 close() 时，它会被缓存以供重用；这就是缓存连接工厂的作用 - 避免为每个请求创建会话的开销。

如果您不希望缓存会话、生产者和消费者的性能优势，请改用 SingleConnectionFactory。请参阅 CachingConnectionFactory 的 JavaDoc。

【讨论】：

感谢您提供的一些有用信息。关闭消费者的原因是它们都有相关 id 选择器，因此它们永远不会成为重用的候选者。鉴于我没有其他消费者，我想我会按照您的建议切换到 SingleConnectionFactory。

但是，您也将丢失会话和生产者的缓存。您之前实际上做了正确的事情 - 使用 cacheConsumers=false 的缓存连接工厂。这样，您就可以两全其美。

【参考方案2】：

使用cachingConnectionFactory 时，以下操作是否有效？

在您的 spring 配置文件中添加连接工厂配置详细信息，如下所示：cacheConsumers="false"

默认行为是 true，这会导致队列中的连接泄漏。