仅当消息数量达到计数或消息在通道中已有一段时间后才从通道中消费答案

【问题标题】：Consume from channel only if the number of messages reaches a count or the message is in the channel since a while仅当消息数量达到计数或消息在通道中已有一段时间后才从通道中消费
【发布时间】：2016-04-27 07:27:25
【问题描述】：

我有一个自定义接收器模块，我希望仅在消息数量达到计数或它们在一段时间内处于通道中时才使用来自 input 的消息。简而言之，我想做一个批量推送。

我尝试在消费后聚合消息数量并将它们存储在由SimpleMessageStore 支持的聚合通道中，并让MessageGroupStoreReaper 检查通道中的消息。

我对这种方法不满意，因为我正在使用消息并将它们存储在内存存储中，我也知道 JDBC 存储，但我不想遵循这种方法作为消息通道spring XD 由 redis/mq 支持，我想根据我的条件从input 频道消费。

我当前的bean配置如下图：

<int:aggregator id="messageAggregator" ref="messageAggregatorBean"
    method="aggregate" input-channel="input" output-channel="aggregatorOutputChannel"
    release-strategy="messageReleaseStrategyBean" release-strategy-method="canRelease"
    send-partial-result-on-expiry="true" message-store="resultMessageStore">
</int:aggregator>

<int:service-activator id="contributionIndexerService"
    ref="contributionIndexerBean" method="bulkIndex" input-channel="aggregatorOutChannel" />

<bean id="resultMessageStore"
    class="org.springframework.integration.store.SimpleMessageStore" />

<bean id="resultMessageStoreReaper"
    class="org.springframework.integration.store.MessageGroupStoreReaper">
    <property name="messageGroupStore" ref="resultMessageStore" />
    <property name="timeout" value="60000" />
</bean>

<task:scheduled-tasks>
    <task:scheduled ref="resultMessageStoreReaper" method="run"
        fixed-rate="10000" />
</task:scheduled-tasks>

有什么想法或方法吗？

提前致谢。

【问题讨论】：

标签： spring spring-integration spring-xd

【解决方案1】：

我不确定您是否能够确定 Broker 队列（Redis/RabbitMQ 甚至普通 JMS）中的消息数量，而不是它们的数量。

你绝对应该消耗他们来做这样的逻辑。

是的，我认为Aggregator 可以帮助您。但是没错：那一定是Persistent Message Store。

案例

如果他们在一段时间内就在频道中

Aggregator 建议使用类似group-timeout 的选项来释放那些尚未达到releaseStrategy 条件的组，但无论如何您都希望在一段时间内释放它们：@987654322 @

【讨论】：