java - Spring 集成: Pass Entire List of object from router to service at one time

Question

我的代码中有以下拆分器 例如

public class CurrencySplitter {

public Collection<CurrencyDomain> split(String[] currency) {
    List<CurrencyDomain> list = new ArrayList<CurrencyDomain>();

    list.add(new CurrencyA(12, "A"));
    list.add(new CurrencyA(13, "A"));
    list.add(new CurrencyB(12, "B"));
    list.add(new CurrencyC(12, "C"));

    return list;
   }
  }

分配器配置:

<si:payload-type-router input-channel="currencyCreateSplitChannel">
    <si:mapping type="com.vce.domain.CurrencyA" channel="createCurrencyAChannel" />
    <si:mapping type="com.vce.domain.CurrencyB" channel="createCurrencyBChannel" />
    <si:mapping type="com.vce.domain.CurrencyC" channel="createCurrencyCChannel" />
</si:payload-type-router>

<si:channel id="createCurrencyAChannel">
</si:channel>
<si:service-activator input-channel="createCurrencyAChannel"
    method="createCurrencyA" output-channel="currencySplitResponseChannel"
    ref="currencyService">
</si:service-activator>

...

服务等级:

public class CurrencyService extends BaseService {

public CurrencyDomain createCurrencyA(List<CurrencyA> currA) {

    System.out.println(currA.size());
    return new CurrencyDao.save(currA);
}

public CurrencyDomain createCurrencyB(CurrencyB currB) {

    System.out.println("B");
    return new CurrencyDomain("B");
}

public CurrencyDomain createCurrencyC(CurrencyC currB) {
    System.out.println("C");
    return new CurrencyDomain("C");
   }

 }

问题是我一次需要整个CurrencyAList，以便我一次保存所有数据。但现在 createCurrenyA 方法被调用两次，并分别调用 dao。

需要有关如何获取所有 currentA 对象作为列表并立即保存的帮助

Answer 1

您需要一个<aggregator>对于之前的情况 <si:service-activator>调用:

拆分器 -> 有效负载类型路由器 -> CurrencyA 的聚合器对象

<si:chain input-channel="createCurrencyAChannel" 
                output-channel="currencySplitResponseChannel">
   <si:aggregator correlation-strategy-expression="1" 
      group-timeout="100" 
      expire-groups-upon-completion="true"
      release-strategy-expression="true"/>
   <si:service-activator method="createCurrencyA" ref="currencyService"/>
</si:chain>

因为您的所有有效负载createCurrencyAChannel是CurrencyA你可以使用简单的<aggregator>配置。这意味着:

• 仅为所有传入消息形成一个组，因为 correlationKey是 1

• 由于我们可能不知道传入消息的数量，因此我们只是通过 timeout 释放该组

• 就我们而言，我们始终应该发布我们的 release-strategy总是true

• 为了允许在上一个版本之后为任何新消息形成一个组，我们应该删除那些 completed组，因此 xpire-groups-upon-completion="true"