Spring集成：在处理所有消息时可以释放的聚合器
我正在构建一个调用许多不同web服务的系统，我希望生成一个关于调用ws后返回的所有错误的报告。
为此，我使用了<代码>
<int:aggregator
correlation-strategy-expression="'${error.msg.correlation.key}'"
input-channel="ws.rsp.error.channel"
output-channel="outboundMailChannel"
ref="errorAggregator"
method="generateErrorReport"
release-strategy-expression="false"
group-timeout="2000"
组在完成时过期=“true”/
<int:service-activator
input-channel="outboundMailChannel"
ref="errorMsgAgregatedActivator"
method="handleMessage"
/>
和激活剂：
@ServiceActivator
public void handleMessage(Message<Collection<Object>> errorList) {
Collection<Object> payload=errorList.getPayload();
System.out.println("error list: "+payload.toString());
}
谢谢
共有1个答案
聚合需要一个适当的释放策略，或者您可以简单地使用释放策略表达式false（从不释放），并在一段时间后使用组超时释放组中的任何内容。
您可能希望使用常量相关性corvery-Strategy-expresision="'foo'"并设置exire-groups-on-补全="true"以便新组从下一条消息开始。
-
我正在尝试使用spring integration设置我的应用程序，作为一名新手，需要以下用例的建议- 有一个队列，来自另一个应用程序的消息将被推送到该队列。我的应用程序使用队列中的消息，进行一些数据处理，然后将其推送到另一个出站队列。目标是以并发方式处理消息。 根据我的理解，我们可以有两种方法- 1.使用#轮询器 2.使用#调度器 从基于轮询器的配置来看，池中似乎有多个可用线程，可以同时获取消息
-
我有一个用例，我的消息被拆分两次，我想聚合所有这些消息。如何才能最好地实现这一点，我应该通过引入不同的序列头来聚合消息两次，还是有办法通过重写消息分组的方法在单个聚合步骤中聚合消息？
-
我正在尝试做一个GroupBy基于共享ID的GeoJSON功能列表，以便通过使用拆分/聚合来聚合这些功能的单个字段，如下所示： 除非我取消对这三行的注释，否则聚合器永远不会发布组，数据库也不会收到任何更新。如果我将groupTimeout设置为小于5秒，则会丢失部分结果。 我预计发布策略默认为，我预计在处理完所有（拆分）功能后会自动释放所有组（REST服务消息中总共只有129个功能）。手动将其设置
-
我们的应用程序中存在以下问题。消息通过入站通道适配器传入，并使用持久消息存储在聚合器中累积。一旦释放策略中定义的条件返回true，消息将被发送到处理的下一阶段。如果在下一个处理阶段抛出异常，事务将回滚，消息将再次放入持久消息存储中。但是，事务不会将消息放回原始队列，因为消息一旦放在聚合器中就会被确认。这不是我们想要的。理想情况下，如果在处理聚合器已批处理的其中一条下游消息时发生异常，则事务只会回滚
-
我们一直在使用SI Kafka进行一个新项目，并取得了很大成功。在最近的一次切换之前，我们使用KafkaTopicOffsetManager来管理我们的消费者主题偏移量。为了避免每个消费者/主题对都有额外的主题，并使用Burrow或lag监控，我们决定使用最新的KafkaNativeOffsetManager，它使用Kafka提供的本机偏移管理。但在切换之后，我们注意到目标主题的消息消耗持续滞后。
-
使用Spring Integr中的拆分器，我拆分了从数据库中的表中选择的数据行。每条消息完成处理后，我想像旧消息一样将每条消息聚合到一条消息中。我该怎么办？我不知道拆分器拆分了多少条消息。我只知道拆分消息头中的相关ID。即使我聚合消息，我也无法制定发布策略。 我如何解决这个问题？ 以及是否有任何方法可以使用jdbc-out站网关或jdbc-out站通道适配器一次插入多行数据，而无需使用拆分器插入每