无法聚合消息，使用错误的方法？答案

【问题标题】：Unable to Aggregate Messages, using wrong approach?无法聚合消息，使用错误的方法？
【发布时间】：2017-03-07 18:12:40
【问题描述】：

我对 Spring Integration 很陌生。我无法弄清楚如何解决以下情况。

我有两个通道为要发送到 REST 端点的最终消息生成两个关键数据。

轮询通道定期生成 OAuth2 令牌并发布（供此通道和其他通道使用）
文件监控通道产生payload（从文件中加载payload的内容）

我需要将通道1的token添加到通道2的header中，类似下图：

                     other channels            __ Ch3 (uses token)
                     subscribed to the -->    /
                     publishing channel    --+ -- Ch4 (uses token)
                                          /   \__ Ch5 (uses token)
Ch1 ==> getToken ==> enrich header & publ.
                                          \    Merge
                                           ==> Token &  ==> Send to endpoint
                                          /    Payload
Ch2 ==>  readFile ==>  create new Payload

我尝试使用聚合器，但不确定关联策略或发布策略使用什么。

我的问题是：

聚合器是否可行？（或者也许还有其他方法？）
关于关联/发布策略的任何建议？

任何方向将不胜感激。

这是我目前所拥有的：

<!-- channels -->
<int:channel id="aggregatedMsgOutChannel"/>

<!-- Token messages are published to here -->
<int:publish-subscribe-channel id="tokenInChannel" />

<!-- Message Aggregator -->
<int:aggregator
        id="messageAndOauthTokenAggregator"
        input-channel="tokenInChannel"
        message-store="simpleMessageStore"
        ref="oauthTokenAggregator"
        method="aggregate"
        output-channel="aggregatedMsgOutChannel">
 </int:aggregator>

 <!-- Define a store for our messages -->
 <bean id="simpleMessageStore" class="org.springframework.integration.store.SimpleMessageStore" />

<bean id="oauthTokenAggregator" class="c.s.i.OauthTokenAggregator">

OauthTokenAggregator.java

@Component
public class OauthTokenAggregator{
    private static final Logger log = LoggerFactory.getLogger(OauthTokenAggregator.class);

    @Aggregator
    public Message aggregate(Collection<Message<?>> messages) {
        log.debug("aggregating...");

        //.... Unsure as to how to fill this section

        return new GenericMessage("test");
    }
}

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

我认为您使用聚合器的方式是正确的。

您应该为生成的令牌和文件内容选择一些公共属性。我很确定它一定在那里，否则你将如何在没有聚合器或 Spring Integration 的情况下连接？
发布策略看起来很简单 - 只是大小为 2 - 令牌和有效负载。
聚合函数实际上只能基于作为有效负载的文件内容生成一条消息 - 从一条消息。并将令牌放入另一条消息的标头中。您可以通过有效负载类型或其中的某些标头来区分消息。

只是我不知道你的业务逻辑的问题，所以，我不能给你好的建议。也许您不应该使用单独的轮询通道适配器，而是使用文件进程生成令牌？..

或者...如果每个文件是单独生成的并且可以在其他地方使用，则每个文件都可以从某个队列中轮询令牌。

这是一个没有聚合器的变体。如果我们仍然坚持那个，我们别无选择，除非选择文件和令牌之间的某种相关性。

【讨论】：

感谢您的建议。让我尝试在文件进程中生成令牌（可能在链中？），然后我会报告我得到了多远；）
您可以通过<enricher> 的子流生成它并按照您的预期填充到标题中：docs.spring.io/spring-integration/reference/html/…

【解决方案2】：

我最终使用了@Artem 建议的方法：在文件进程中生成令牌。

主要实现的功劳来自article in DZone。为了完整起见，在此处复制代码（完全归功于文章作者）。

所以，基本上，

我在第一个 chain 中启动了文件进程，将其签入到 claim-check-in 并将其 UUID 放入我的标题中。
而不是在中间使用这个chain

<int:service-activator expression="new String('different string')"/>

我去拿了一个令牌并将其添加到我的标题中并

继续到最后一个chain，从声明检查中获取我的有效负载并将其发布到 REST 端点

<int:chain input-channel="claim-check-in-channel"
           output-channel="processing-channel">
    <int:claim-check-in message-store="simpleMessageStore"/>
    <int:header-enricher>
        <int:header 

            name="#{T(com.l8mdv.sample.ClaimCheckGateway).CLAIM_CHECK_ID}"
            expression="payload"/>
    </int:header-enricher>
</int:chain>

<int:chain input-channel="processing-channel"
           output-channel="claim-check-out-channel">
    <int:service-activator expression="new String('different string')"/>
</int:chain>

<int:chain input-channel="claim-check-out-channel">
    <int:transformer
            expression="headers.get('#{T(com.l8mdv.sample.ClaimCheckGateway)
            .CLAIM_CHECK_ID}')"/>
    <int:claim-check-out message-store="simpleMessageStore"
                         remove-message="true"/>
</int:chain>

【讨论】：