Spring Integration+ActiveMQ并行处理问题：消息发送报错及消费需求

一、解决消息生产者的错误 #

你遇到的IllegalStateException核心原因是Spring Integration找不到匹配的消息处理方法：你的QkaapzzMessageProducer.sendMessage()定义了两个参数，但Spring Integration默认只会将消息本身作为单个参数传递给处理方法，没有第二个参数queue2的来源，因此无法匹配到合适的方法。

修正方案： #

1. 简化自定义Producer方法（快速修复）

去掉不必要的queue2参数，或者通过消息头传递队列名称。另外不建议手动管理JMS的Connection、Session和MessageProducer——Spring Integration已经封装了成熟的JMS组件，能自动帮你处理资源管理，避免手动编码的风险。

修改后的Producer类示例：

public class QkaapzzMessageProducer {
    public static final String HELLO_WORLD_QUEUE = "testQueue";

    // 仅接收Message<?>类型的参数，匹配Spring Integration的调用逻辑
    public Message<?> sendMessage(Message<?> msg) throws JMSException {
        // 这里可以保留你的业务逻辑，后续也可以用Spring集成组件替代
        return msg;
    }
}

2. 改用Spring Integration JMS Outbound Adapter（最佳实践）

完全替换自定义Producer，使用Spring提供的<jms:outbound-channel-adapter>来发送消息，这是Spring集成JMS的标准方式：

<!-- 替换原来的service-activator配置 -->
<jms:outbound-channel-adapter id="jmsOutbound" 
                               channel="postChannel" 
                               destination="testQueue"
                               connection-factory="jmsConnectionFactory"/>

这样就不需要自己编写QkaapzzMessageProducer类，Spring会自动处理JMS连接、会话和消息发送的全流程。

3. 动态指定队列名称（如果需要）

如果必须根据消息内容动态选择队列，可以通过destination-expression从消息头或负载中获取队列名：

<jms:outbound-channel-adapter id="jmsOutbound" 
                               channel="postChannel" 
                               destination-expression="headers['targetQueue'] ?: 'testQueue'"
                               connection-factory="jmsConnectionFactory"/>

此时只需在发送消息时设置targetQueue头即可，无需自定义Producer。

二、确保并行处理的正确性 #

你的<int:publish-subscribe-channel id="postChannel"/>配置是正确的，它会将消息广播给所有订阅该通道的处理器（JDBC Outbound Adapter和JMS Outbound Adapter），实现数据并行存储到数据库和发送到MQ的需求。

如果需要真正的异步并行执行，可以给通道添加任务执行器：

<int:publish-subscribe-channel id="postChannel">
    <int:dispatcher task-executor="taskExecutor"/>
</int:publish-subscribe-channel>

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="5"/>
    <property name="maxPoolSize" value="10"/>
</bean>

三、实现ActiveMQ消息消费者 #

要接收队列中的消息，推荐使用以下两种Spring Integration配置：

1. 消息驱动消费者（实时监听） #

这种方式会持续监听队列，有消息就立即处理：

<!-- 定义消费者处理通道 -->
<int:channel id="consumerChannel"/>

<!-- JMS消息驱动适配器，实时监听队列 -->
<jms:message-driven-channel-adapter id="jmsConsumerAdapter"
                                    destination="testQueue"
                                    connection-factory="jmsConnectionFactory"
                                    channel="consumerChannel"/>

<!-- 处理消息的业务类 -->
<int:service-activator id="messageConsumer"
                       input-channel="consumerChannel"
                       ref="consumerService"
                       method="handleMessage"/>

对应的消费者服务类：

public class ConsumerService {
    public void handleMessage(Message<?> msg) {
        // 这里编写消息处理逻辑，比如打印、存储到数据库等
        System.out.println("Received message: " + msg.getPayload());
    }
}

2. 轮询式消费者（定时拉取） #

如果不需要实时监听，可以用轮询方式定期拉取消息：

<jms:inbound-channel-adapter id="jmsPollingAdapter"
                             destination="testQueue"
                             channel="consumerChannel"
                             connection-factory="jmsConnectionFactory">
    <int:poller fixed-delay="5000"/> <!-- 每5秒拉取一次消息 -->
</jms:inbound-channel-adapter>

四、其他配置注意事项 #

• 你的JDBC Outbound Adapter中，parameterExpressions定义了lstName但SQL语句未使用，建议删除或补充到SQL中，避免无效配置。
• 确保jmsConnectionFactory、dataSource等核心Bean已正确配置，能正常连接到ActiveMQ和数据库。

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