如何通过Spring Integration调用消息生产者并并行执行数据库插入与MQ发送

嘿，看起来你需要在收到POST请求后，同时并行搞定数据库插入和ActiveMQ消息推送这俩事儿。基于你给的现有配置，咱们可以用Spring Integration的**发布-订阅通道（Publish-Subscribe Channel）**结合线程池来实现并行处理，具体改造方案如下：

核心思路 #

Spring Integration的publish-subscribe-channel能把同一条消息分发给多个消费者，再配合task-executor线程池，就能让多个消费者在独立线程里并行干活。咱们只需要把原来的postChannel改成发布订阅类型，然后把JDBC出站适配器和新增的JMS出站适配器都挂到这个通道上就行。

具体配置修改 #

1. 把原postChannel换成带线程池的发布订阅通道 #

把原来的单消费者通道<int:channel id="postChannel" />替换成下面的配置：

<int:publish-subscribe-channel id="postChannel">
    <!-- 配置线程池实现并行执行，参数可以根据你的业务压力调整 -->
    <int:dispatcher task-executor="taskExecutor" />
</int:publish-subscribe-channel>

<!-- 定义线程池Bean，按需调整核心线程数、最大线程数等参数 -->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="5" />
    <property name="maxPoolSize" value="10" />
    <property name="queueCapacity" value="25" />
</bean>

这样postChannel收到消息后，会同时把消息发给所有注册的消费者，而且每个消费者在单独的线程里执行，完美实现并行。

2. 添加JMS出站适配器 #

在现有配置里加上这个JMS出站适配器，专门用来把消息发送到ActiveMQ队列：

<!-- JMS出站适配器：把收到的消息发送到指定ActiveMQ队列 -->
<int-jms:outbound-channel-adapter id="jmsOutboundAdapter"
                                  channel="postChannel"
                                  connection-factory="jmsConnectionFactory"
                                  destination="helloWorldJMSQueue" />

这里的destination可以换成你实际要用的队列名称，如果需要把表单数据转换成特定格式（比如JSON）的JMS消息，可以加个message-converter属性指定自定义的转换器。

3. 保留原有JDBC出站适配器 #

你原来的JDBC出站适配器不用大改，它会自动成为postChannel的另一个消费者，和JMS适配器一起并行执行：

<int-jdbc:outbound-channel-adapter query="insert into user_registration (USER_FSTNAME,USER_ADDRESS,USER_STATE,USER_CITY,USER_OCCUPATION,USER_EMAIL,USER_CONTACT,USER_PASSWORD) values (:fstName,:addrss,:state,:city,:occupation,:email,:contact,:password)" 
                                  channel="postChannel" 
                                  data-source="dataSource" 
                                  id="sample" 
                                  sql-parameter-source-factory="spelSource" />

关键注意点 #

• 并行性保证：靠publish-subscribe-channel的task-executor，数据库插入和消息发送会在不同线程里同时跑，互相不阻塞。
• 事务一致性：如果要求两个操作必须同时成功或失败（分布式事务），那得配置JTA事务管理器（比如Atomikos），但这样会牺牲并行性；如果允许最终一致，当前方案就足够用了。
• 消息格式转换：要是JMS需要特定格式的消息（比如JSON），可以在JMS适配器前面加个<int:object-to-json-transformer>，或者自定义MessageConverter来处理。

整合后的完整配置示例 #

<!-- 线程池配置，按需调整参数 -->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="5" />
    <property name="maxPoolSize" value="10" />
    <property name="queueCapacity" value="25" />
</bean>

<!-- 发布订阅通道，实现消息并行分发 -->
<int:publish-subscribe-channel id="postChannel">
    <int:dispatcher task-executor="taskExecutor" />
</int:publish-subscribe-channel>

<int:channel id="requestChannel" />
<int:channel id="outputChannel" />
<int:channel id="errorChannel" />

<!-- GET请求的HTTP入站网关 -->
<int-http:inbound-gateway request-channel="requestChannel" reply-channel="outputChannel" supported-methods="GET" path="/register" view-name="register">
    <int-http:request-mapping />
</int-http:inbound-gateway>

<!-- POST请求的HTTP入站网关 -->
<int-http:inbound-gateway request-channel="postChannel" reply-channel="outputChannel" supported-methods="POST" path="/registerNew" error-channel="errorChannel" view-name="login">
</int-http:inbound-gateway>

<!-- JDBC出站适配器：处理数据库插入 -->
<int-jdbc:outbound-channel-adapter query="insert into user_registration (USER_FSTNAME,USER_ADDRESS,USER_STATE,USER_CITY,USER_OCCUPATION,USER_EMAIL,USER_CONTACT,USER_PASSWORD) values (:fstName,:addrss,:state,:city,:occupation,:email,:contact,:password)" 
                                  channel="postChannel" 
                                  data-source="dataSource" 
                                  id="sample" 
                                  sql-parameter-source-factory="spelSource" />

<!-- JMS出站适配器：发送消息到ActiveMQ -->
<int-jms:outbound-channel-adapter id="jmsOutboundAdapter"
                                  channel="postChannel"
                                  connection-factory="jmsConnectionFactory"
                                  destination="helloWorldJMSQueue" />

<!-- 服务激活器 -->
<int:service-activator ref="userService" input-channel="requestChannel" output-channel="outputChannel" method="message"/>

<!-- SQL参数源工厂 -->
<bean id="spelSource" class="org.springframework.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
    <property name="parameterExpressions">
        <map>
            <entry key="fstName" value="payload[firstName]" />
            <entry key="addrss" value="payload[address]" />
            <entry key="state" value="payload[state]" />
            <entry key="city" value="payload[city]" />
            <entry key="occupation" value="payload[occupation]" />
            <entry key="email" value="payload[email]"/>
            <entry key="contact" value="payload[contact]"/>
            <entry key="password" value="payload[password]"/>
        </map>
    </property>
</bean>

<!-- 原有的JMS消息驱动适配器（如果需要保留队列消费逻辑） -->
<jms:message-driven-channel-adapter id="helloWorldJMSAdapater" destination="helloWorldJMSQueue" connection-factory="jmsConnectionFactory" channel="postChannel" />

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