Spring集成轮询聚合器（以编程方式）

Question

我一直在使用Spring boot，并且在我的项目中删除了所有XML文件。不幸的是，它还使用了Spring集成，根据我的经验，Spring集成非常基于XML。

我有一个场景，要求我有一个聚合器，并让聚合器每隔x秒轮询一次。

这可以像这样使用XML来完成（从以前的SO问题中获得的示例）：

<!-- 
    the poller will process 100 messages every minute 
    if the size of the group is 100 (the poll reached the max messages) or 60 seconds time out (poll has less than 100 messages) then the payload with the list of messages is passed to defined output channel
-->
<int:aggregator input-channel="logEntryChannel" output-channel="logEntryAggrChannel"
    send-partial-result-on-expiry="true"
    group-timeout="60000"
    correlation-strategy-expression="T(Thread).currentThread().id"
    release-strategy-expression="size() == 100">
    <int:poller max-messages-per-poll="100" fixed-rate="60000"/>
</int:aggregator>

我已经找到了一个类，这个类有点像是在玩把戏，它的bean定义是：

@Bean(name = "aggregatingMessageHandler")
public AggregatingMessageHandler aggregatingMessageHandler() {

    AggregatingMessageHandler aggregatingMessageHandler =
            new AggregatingMessageHandler(messageGroupProcessorBean(),
                    new SimpleMessageStore(10));

 aggregatingMessageHandler.setCorrelationStrategy(customCorrelationStrategyBean());

    aggregatingMessageHandler.setReleaseStrategy(
            new TimeoutCountSequenceSizeReleaseStrategy(3,
                    TimeoutCountSequenceSizeReleaseStrategy.DEFAULT_TIMEOUT));

    aggregatingMessageHandler.setExpireGroupsUponCompletion(true);

    aggregatingMessageHandler.setOutputChannel(outputAggregatedChannelBean());

    return aggregatingMessageHandler;
}

但是，只有在与此处理程序关联的inboundChannel中接收到新消息时，才触发ReleaseStrategy的canRelease（）方法，而不是在固定的时间间隔内（这不是期望的结果）。我希望将超过一分钟的所有组重定向到输出通道。我的问题是——有没有一种方法可以通过编程方式附加一个轮询器，比如XML定义中的轮询器？

Answer 1

对于Java

Aggregator组件具有AggregatorFactoryBean，以便更容易Java配置。

无论如何，您必须注意，在处理程序定义上有一个@ServiceActivator注释和一个@Bean。而且，@ServiceActivator具有轮询器属性。

还要注意，Spring集成有JavaDSL。

你问题的另一部分是有点困惑。轮询器完全与发布策略无关。在这种情况下，它负责从PollableChannel接收消息，即logEntryChannel。并且只有在这之后，已经轮询的消息才会被放置到聚合器中，以便进行关联和发布逻辑。

该示例中所做的是完全不同的故事，我们可以在单独的SO线程中讨论它。