Spring MDP：轮询间隔？

Question

一位客户使用这种模式：

• Apache Camel 和 CXF JMS 接收器
• 这些内部使用 Spring MDP（消息驱动 POJO）来实现其消息接收器
• 它们部署在 IBM WebSphere Application Server 7 上
• 队列管理器是 IBM Websphere MQ 6
• Spring MDP 使用 JNDI 队列连接工厂绑定到队列管理器 - 支持连接池和会话池

这是一个这样的消息接收器的例子，这个是使用骆驼的：

<bean id="ibmmq" class="org.apache.camel.component.jms.JmsComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

<!-- JNDI reference to the queue manager -->
<jee:jndi-lookup id="myTargetConnectionFactory" jndi-name="${mq.queueconnectionfactory}"/>

<bean id="jmsDestResolver" class="org.springframework.jms.support.destination.JndiDestinationResolver"/>

<bean id="myConnectionFactory" class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter">
    <property name="targetConnectionFactory" ref="myTargetConnectionFactory"/>
    <property name="username" value="SOME_USER"/>
    <property name="password" value=""/>
</bean>

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">

    <property name="connectionFactory" ref="${mq.connectionfactorybean}" />        
    <property name="destinationResolver" ref="jmsDestResolver" />        
    <property name="concurrentConsumers" value="1" />
    <property name="maxConcurrentConsumers" value="1" />

    <!--
        NOTE: If we try to use a cache without a transactionManager we get "Connection closed" errors
    -->
    <property name="cacheLevelName" value="CACHE_NONE" />
</bean>

问题：WebSphere MQ 管理员报告了针对队列管理器的大量 MGET() 请求。目前的假设是那些接收者不断地轮询频道以获取新消息。

他们似乎对 MDB（消息驱动 bean）没有这个问题。 MDP 异步实现真的是一种轮询机制吗？如果是这样，有没有办法限制队列管理器的行程？也许增加轮询间隔？任何见解将不胜感激。

Answer 1

我不确定 CXF，但对于 Camel 听众：

JmsConfiguration 中的默认 JMS 使用者似乎是“默认”类型。 意思是，它将实现一个来自 Spring 的 DefaultMessageListenerContainer。

来自Javadoc

使用普通 JMS 客户端 API 的消息侦听器容器变体，特别是 MessageConsumer.receive() 调用循环

接收调用将映射到 MQ GET 调用。

还可以选择指定Simple 类型的消费者，我猜这就是您想要的。

使用普通 JMS 客户端 API 的 MessageConsumer.setMessageListener() 的消息侦听器容器

我不确定，但 Spring 文档指出简单消息侦听器容器不支持 XA 事务。这可能需要考虑，因为您在应用程序服务器中运行。

Answer 2

我们的 Mainframe Q 也有类似的问题。IBM 大型机 q：

注意应用进程ID作为默认用户 要传递给队列管理器的身份。如果应用程序是 以客户端传输模式运行，则此进程 ID 必须与 服务器机器上的相关授权。如果不同 身份是必需的，那么应用程序应该使用 createConnection(username, password) 方法。

或者换句话说，IBM 使用 JVM 进程 ID 登录到 mq，除非我们发送适当的凭据。我们使用的是 Spring，所以每次 DefaultMessageListenerContainer 轮询 Q 时，它都必须与它一起发送凭据。我连接了其中一个婴儿和 Bam，像一个魅力一样工作：

public class CustomConnectionFactory extends CachingConnectionFactory {

    private String username;
    private String password;

    ...

    /**
     * This is the secret sauce. Each time when we make a connection, we send
     * the username/password.
     */
    protected Connection doCreateConnection() throws JMSException {
        return getTargetConnectionFactory().createConnection(this.username, this.password);
    }

我们的大型机更快乐。后来我们切换到分布式 MQ，一切都变得更好了！

这是我们的最终设置：

<!-- This hooks us up to the jndi -->
<jee:jndi-lookup id="telematicsJNDIConnectionFactory" jndi-name="${mq.jndi}" cache="true" lookup-on-startup="true" />
<!-- The outer wrapper must be TransactionAware, the inner custom one will cache the connection -->
<bean id="telematicsConnectionFactory" class="org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy">
    <property name="targetConnectionFactory">
        <bean class="cat.dds.tmatic.utils.CustomConnectionFactory">
            <property name="targetConnectionFactory">
                <ref bean="telematicsJNDIConnectionFactory" />
            </property>
            <property name="username">
                <value>${mq.user}</value>
            </property>
            <property name="password">
                <value>${mq.pass}</value>
            </property>
            <property name="sessionCacheSize">
                <value>10</value>
            </property>
        </bean>
    </property>
    <property name="synchedLocalTransactionAllowed" value="true" />
</bean>