弹簧积分并行富集.初始线程丢失

Question

我们有以下春季集成流程：

jms消息驱动的通道适配器-> . ->发布/子通道-> 3丰富了订阅->聚合器-> .

每个浓缩器都指定了task-scheduler，因此它们是并行工作的。

不幸的是，这种方法不能很好地工作，因为原始JMS线程丢失了。

我希望jms-message-driver-channel-adapter和aggregator在同一个线程中运行，但是aggregator (和下面的处理程序)运行在“最后”中丰富了她的线程。

我怎样才能做到这一点？我在spring-int文档中没有看到这篇文章。

在Gary的回复之后添加

我决定以更自然的方式来实现这一点：

<int:service-activator method="enrich" input-channel="in" output-channel="out">
  <bean class="com.xxx.ParallelEnricher" p:timeoutMs="10000">
    <constructor-arg ref="taskExecutor" />
    <constructor-arg>
      <list>
        <bean class="com.xxx.Enricher1" />
        <bean class="com.xxx.Enricher2" />
        <bean class="com.xxx.Enricher3" />
      </list>
    </constructor-arg>
  </bean>
</int:service-activator>

其中，ParallelEnricher是一个可重用的类，它为每个用户调用“未来的taskExecutor.submit(Runnable)”，并处理超时。也许我遗漏了一些东西，但是最好在下面这样的消息上配置并行操作：

<int:service-activator method="enrich" input-channel="in" output-channel="out"
                       timeout="10000" task-executor="taskExecutor">
  <list>
    <bean class="com.xxx.Enricher1" />
    <bean class="com.xxx.Enricher2" />
    <bean class="com.xxx.Enricher3" />
  </list>
</int:service-activator>

Answer 1

一种解决方案是添加网关中间流；这样，JMS线程将等待最终使用者的答复--它必须返回“某个东西”(只要关闭他的output-channel，框架将返回给网关，在那里它可以被丢弃).

<int:service-activator input-channel="fromJMS" output-channel="nullChannel" 
    ref="gw" />

<int:gateway default-request-channel="myPubSub" />

只需确保最终的使用者(大概是在聚合器之后的某个地方)发送一个答复(不管它是什么；它将通过发送到nullChannel而被丢弃)。

注意，默认情况下，线程将无限期地等待；如果需要超时并回滚消息，则需要额外的逻辑。

另一种解决方案是添加第四个订阅服务器，这是receive()从QueueChannel获得的一个简单服务。同样，最终的使用者会发送一条消息来触发发布。您需要在pub/sub之前添加一个新的QueueChannel报头(<int:header name="myReleaseTriggerChannel" expression="new ...QueueChannel()"/>；最终的使用者向该消息头发送“东西”)(可以是一个简单的路由器)。

这个解决方案需要少量的用户代码(从队列接收)，但是以这种方式处理超时更容易。