Spring Integration-JPA入站通道适配器、拆分器、服务激活器并发工作单元

Question

我有一个非常简单的Spring集成管道设置

1. 用JPA将n次状态为X的资产加载到通道中
2. 将每个JPA实体拆分到一个新通道上
3. 使用服务激活器处理JPA实体，最后将实体的状态更新为y

这个管道的同步特性意味着JPA入站通道适配器只在上一个入站通道适配器中的所有消息被处理并发送到nullchannel之后才触发

这里的问题是service activator处理一条消息需要大约1秒（大部分时间是对REST API的调用），因此，处理250个JPA实体的队列可能需要250秒。

如果我们同时调用REST API，比如说5次，它仍然需要1秒。

因此，我想知道是否可以对我们的管道进行简单的修改，比如添加一个聚合器和一个任务执行器，这将允许整个管道作为一个同步的“工作单元”运行，但允许服务激活器并发处理。

<channel id="newAssetChannel" />
<channel id="splitAssetChannel" />

<int-jpa:inbound-channel-adapter
        id="newAssetChannelAdapter"
        channel="newAssetChannel"
        entity-manager-factory="entityManagerFactory" 
        entity-class="com.foo.domain.Asset" 
        jpa-query="select a from Asset a where (a.status = 'NEW' or a.status = 'UPDATED') and a.health = 'OK' ORDER BY a.priority DESC, a.updatedDate ASC"
        max-results="250">
    <poller fixed-rate="5000" max-messages-per-poll="1" />
</int-jpa:inbound-channel-adapter>

<splitter expression="payload"
    input-channel="newAssetChannel"
    output-channel="splitNewAssetChannel" />

<service-activator
    id="newAssetServiceActivator"
    input-channel="splitNewAssetChannel"
    output-channel="nullChannel"
    ref="assetProcessor"
    method="processNew" />

Answer 1

嗯，聚合器确实是等待所有答复的正确方法，但是与拆分器之后的executorchannel一起，还是可以解放poller的双手。所以，在并行拆分-聚合之前-之后应该有一些障碍。

您可以使用 :

<gateway id="gateway" default-request-channel="splitterChannel"/>

<service-activator id="gatewayTestService" input-channel="newAssetChannel" output-channel="saveRowsChannel" ref="gateway"/>

拆分器的output-channel必须是executorchannel。NewAssetServiceActivator必须输出到聚合器。聚合器没有输出通道，这意味着对网关的答复。