Spring集成 - 用于整理/批处理服务调用的适当模式

Question

当有特定事件发生时,我正在调用一个远程服务来加载产品的定价数据.一旦加载,产品定价就会被广播,供另一个消费者在其他地方处理.

我不想在每个事件上调用远程服务,而是将事件批量分组,然后一次性发送.

我基于聚合器拼凑了以下模式.虽然它有效但很多都有"闻起来" - 特别是我的SimpleCollatingAggregator.我是Spring Integration和EIP的新手,并且怀疑我是在滥用组件.

代码

通过调用下面的方法,我的代码在代码的其他地方触发@Gateway:

public interface ProductPricingGateway {    
    @Gateway(requestChannel="product.pricing.outbound.requests")
    public void broadcastPricing(ProductIdentifer productIdentifier);
}

然后将其连接到聚合器,如下所示:

<int:channel id="product.pricing.outbound.requests" />
<int:channel id="product.pricing.outbound.requests.batch" />
<int:aggregator input-channel="product.pricing.outbound.requests"
output-channel="product.pricing.outbound.requests.batch" release-strategy="releaseStrategy"
    ref="collatingAggregator" method="collate"
    correlation-strategy-expression="0"
    expire-groups-upon-completion="true" 
    send-partial-result-on-expiry="true"/>  
<bean id="collatingAggregator" class="com.mangofactory.pricing.SimpleCollatingAggregator" />
<bean id="releaseStrategy" class="org.springframework.integration.aggregator.TimeoutCountSequenceSizeReleaseStrategy">
    <!-- Release when: 10 Messages ... or ... -->
    <constructor-arg index="0" value="10" />
    <!-- ... 5 seconds since first request -->
    <constructor-arg index="1" value="5000" />
</bean>

这是聚合器实现:

public class SimpleCollatingAggregator {

    public List<?> collate(List<?> input)
    {
        return input;
    }

}

最后,这将消耗在以下内容@ServiceActivator:

@ServiceActivator(inputChannel="product.pricing.outbound.requests.batch")
public void fetchPricing(List<ProductIdentifer> identifiers)
{
        // omitted
}

注意:在实践中,我也使用@Async,以使调用代码尽可能快地返回.我也有很多问题,我将转向一个单独的问题.

问题1: 鉴于我想要实现的目标,聚合器模式在这里是一个合适的选择吗？感觉就像很多样板 - 是否有更好的方法？

问题2: 我正在使用固定的校对值0,有效地说:"如何对这些消息进行分组并不重要,随着它们的到来就把它们带走."

这是实现这一目标的适当方式吗？

问题3: SimpleCollatingAggregator我看起来不对劲.

我希望这可以接收我的个人入站ProductIdentifier对象,并将它们分组,然后传递它们.这有效,但它是否合适？是否有更好的方法来实现同样的目标？

Answer 1

Q1：是的，但请参阅 Q3 和下面的进一步讨论。
Q2：这是说“不需要相关性”的正确方式（但你需要expire-groups-on-completion你拥有的）。
Q3：在这种情况下，您不需要自定义聚合器，只需使用默认的聚合器（删除 ref 和 method 属性）。

请注意，聚合器是一个无源组件；释放是由新消息的到达触发的；因此，释放策略的第二部分只会在新消息到达时才会启动（5 秒后不会自发释放该组）。

但是，您可以MessageGroupStoreReaper为此目的配置： http://static.springsource.org/spring-integration/reference/html/messaging-routing-chapter.html#aggregator