AMQP Camel中JMS MessageConsumer每秒关闭问题的排查与解决建议

从日志中的关键错误信息Received disposition with role=receiver delivery-id=128 settled=true: unsupported outcome=modified [condition = amqp:not-implemented可以明确，问题核心是Broker与Qpid JMS客户端之间的AMQP协议处置结果不兼容——Broker发送了客户端不支持的modified类型消息处置指令，直接触发了MessageConsumer关闭并尝试重建。

以下是针对性的解决思路及配置调整方案：

1. 禁用AMQP Modified处置支持 #

修改连接工厂配置，添加参数禁用客户端对modified处置的支持，避免因协议不兼容导致消费者关闭：

<bean class="org.messaginghub.pooled.jms.JmsPoolConnectionFactory" id="test-connection-pool">
    <property name="connectionFactory" ref="test-connection-factory"/>
    <property name="maxConnections" value="2"/>
    <property name="maxSessionsPerConnection" value="1000"/>
    <!-- 添加禁用modified处置的属性 -->
    <property name="connectionFactoryProperties">
        <map>
            <entry key="amqp.disable.modified" value="true"/>
        </map>
    </property>
</bean>

也可以直接在Camel路由的AMQP URI中指定该参数：

<from uri="amqp:queue:Q/A?concurrentConsumers=10&amp;maxConcurrentConsumers=10&amp;connectionFactory.amqp.disable.modified=true"/>

2. 调整消息确认模式适配批量场景 #

当前Queue A每300毫秒产生200条消息，属于高吞吐量场景，默认的自动确认模式可能因频繁ACK操作引发连接异常。建议调整为DUPS_OK_ACKNOWLEDGE模式，减少确认开销：

<from uri="amqp:queue:Q/A?concurrentConsumers=10&amp;maxConcurrentConsumers=10&amp;acknowledgementModeName=DUPS_OK_ACKNOWLEDGE"/>

3. 优化连接池与会话池配置 #

当前连接池maxConnections=2、maxSessionsPerConnection=1000，虽然会话数足够支撑10个并发消费者，但可以添加池的复用参数，避免会话频繁创建销毁：

<bean class="org.messaginghub.pooled.jms.JmsPoolConnectionFactory" id="test-connection-pool">
    <property name="connectionFactory" ref="test-connection-factory"/>
    <property name="maxConnections" value="2"/>
    <property name="maxSessionsPerConnection" value="1000"/>
    <property name="connectionFactoryProperties">
        <map>
            <entry key="amqp.disable.modified" value="true"/>
        </map>
    </property>
    <!-- 添加池复用配置 -->
    <property name="blockIfSessionPoolIsFull" value="true"/>
    <property name="idleTimeout" value="30000"/>
</bean>

4. 调整JMS监听器的恢复策略 #

日志显示DefaultJmsMessageListenerContainer在自动尝试恢复，但可以调整恢复间隔，避免短时间内频繁重建消费者：
在AMQP URI中添加recoveryInterval参数：

<from uri="amqp:queue:Q/A?concurrentConsumers=10&amp;maxConcurrentConsumers=10&amp;recoveryInterval=5000"/>

5. 确认Broker与客户端的AMQP版本兼容性 #

检查Broker（如RabbitMQ、ActiveMQ Artemis等）的AMQP协议版本，确保与Qpid JMS客户端版本匹配。例如，若Broker启用了AMQP 1.0以外的版本，需在客户端配置中指定对应协议版本：

<property name="connectionFactoryProperties">
    <map>
        <entry key="amqp.version" value="1.0"/>
        <entry key="amqp.disable.modified" value="true"/>
    </map>
</property>

内容的提问来源于stack exchange，提问作者Jack