问题标签 [aggregator]

我是使用 Apache Camel 的新手。你能帮我移动文件吗？我有这样的路线：

因此，应将有错误的文件移至“失败”目录，将没有错误的文件移至“成功”目录。我尝试在聚合所需消息后生成异常（在解析有错误的文件时），以便将文件移动到目录“失败”，但所有文件都移动到目录“成功”，即使出现异常。

如果我在聚合器之前抛出异常，文件移动到“失败”目录但最后一个“到”（发送邮件）不起作用。

我是 Drupal 新手。我需要显示一个包含 4 个 RSS 提要项目的列表，上面有一个选择菜单来更改提要并更新列表（通过 ajax）。我已经对 Views 模块进行了一些调查，但我不知道它是否是我需要的。有人可以指出我正确的方向吗？我应该尝试使用 jQuery 来做到这一点吗？

这个问题与SI论坛中的这篇帖子有关，但是由于论坛关闭，我在这里发布它以继续线程：

http://forum.spring.io/forum/spring-projects/integration/748192-messages-not-flowing-when-using-jms-channels

总而言之，我有一个带有 Redis 消息存储的聚合器和一个每 60 秒安排一次的收割机。使用 JMS 通道将消息发送到聚合器。这是配置：

我对线程转储有点陌生，但据我所知，DefaultMessageListener 线程被启动 MessageReaper 的 taskScheduler 卡住了。特别是在 ReentrantLock 类中。

任何想法？也许我必须做一些其他的配置来避免这种情况

任何帮助表示赞赏

提前致谢！古兹曼

这篇文章与这个有关

使用 Aggregator + MessageStoreReaper + Redis 的 Spring 集成死锁？

但这条消息太长，无法发布。我继续原帖

我升级到最新的 Java 7 版本 1.7.0_60-b19 但问题仍然存在。

我做了另一个线程转储，发现了同样的问题：所有 DefaultMessageListenerContainers（计数 20）都被 AbstractCorrelatingMessageHandler 锁调用中的 taskScheduler（entityScheduler-3）锁定。

这是调度程序和聚合器配置：

这是线程转储部分：

是否有可能由于某种原因 AbstractCorrelatingMessageHandler 的 handleMessageInternal 方法没有解锁？我使用 Redis 消息存储并看到 MsgStoreReaper 也使用了锁。这可能是个问题吗？

非常感谢您在这里的帮助！问候古兹曼

更新：

我在日志中发现了这条消息：

但没有例外，也没有堆栈跟踪。TaskUtils 错误对您有什么影响吗？

我有这个聚合器来删除重复文件：

此查询按文档分数排序。但是，我想按 _timestamp 字段排序。这是可能的 ？我用日期直方图聚合器进行了测试。但没有成功。

我有以下工作流程。

1. 入站通道

2. 分离器

3. 拆分通道的任务执行器 - 所有线程执行相同的工作流。

3.a. 构造请求

3.b。网关消息端点的服务激活器包装器。

3.c。带有错误通道配置的 http-outbound-gateway 上的网关包装器（在调用 http-outbound-gateway 时处理异常）

3.d。http-出站网关

4. 聚合器

5. spring 集成工作流中的响应。

如果在 3.d 中发生异常，则控制转到为网关错误通道配置的服务激活器。我将以下内容从失败的消息复制到新标头到传递给错误通道的标头。

一个。相关性Id b. 序号 C. 序列大小

但是在聚合拆分器响应时，DefaultAggregatingMessageGroupProcessor.java 会删除冲突的标头，并在将控制权提供给聚合器之前删除错误通道和回复通道。

因此，一旦聚合器完成它的操作，它就无法找到回复或错误通道，并导致异常。

我正在使用 spring-integration-core 版本 2.2.1，但我不确定为什么在标头聚合期间会删除回复通道和错误通道。

任何有关解决此问题的意见都会有很大帮助。

谢谢你 ：）

编辑 1： 非常感谢 Gary 帮助我解决这个问题。我正在分享我当前的配置

我注意到的是，通过网关时的每个拆分通道都会为错误和回复创建一个新的临时通道，并且在从网关返回响应后，它会保留保留的（原始入站）错误和回复通道标头。正如您所提到的，在控制到达错误转换器之后，保留保留的标头的流程被破坏，并且聚合消息组处理器接收到三个不同的临时通道实例，因此将它们删除。我计划拥有一个自定义消息组处理器并修改用于聚合标头的冲突解决策略并提出此配置。

不过，我还没有对此进行测试。但是基于这个讨论，这看起来不是一个可行的解决方案。

看起来我在网关中的错误处理配置不正确。但是，我对您的这句话感到困惑“与其直接转发消息，不如简单地处理错误流中的错误并将结果正常返回给网关“包装器””。如果我删除错误通道，当异常发生时我将如何恢复控制？可能是我想在这里理解一些东西。你能详细说明一下吗？

我想计算 Elasticsearch 中的术语聚合器计数之间的 pourcentile。

我的查询：

结果聚合器：

我想要“term1”和“term4”之间的这个指标示例：56%

我的业务逻辑需要能够更改聚合器上配置的组超时。

代码如下所示：

问题是：

• 我想向用户显示当前值，但是...
• 吸气剂受到保护。

可能的解决方案：

• 我应该在这个bean上注入一个表达式然后改变它的值吗？
  • 意思是，它会在下次构建消息组时用新结果重新评估表达式？
• 我应该使用我自己的具有公共设置器的处理程序来扩展 AggregatingMessageHandler 吗？
• 或者这是一个错误，应该在下一个版本中修复？

根据我自己的实验（在任何地方都找不到此文档），如果 2 条消息具有相同的相关 ID 和序列号，聚合器将只获取第一条消息并丢弃/忽略另一条消息。

有没有办法让聚合器使用收到的最后一条消息？

聚合会将有效负载合并为 1 个字符串。

简单场景：3条相关ID相同，序列大小为2的消息，按接收时间排序

• 序列号：1；有效载荷：abc
• 序列号：1；有效载荷：定义
• 序列#：2；有效载荷：ghi

当前输出：abcghi

预期输出：defghi

这种情况发生在序列# 2 缺失时，这意味着第一条消息。并且相关ID（从解码的有效载荷中获得）非常有限，因此会被多次使用。

原始信息

原始消息采用以下格式：

• "序列大小","序列号","ID","文本"
• ID 范围在 0-9 之间
• 示例消息：2,1,8,abc

示例原始消息有效负载：

• 2,1,8,abc
• 2,1,8,定义
• 2,2,8,吉

聚合器基本上结合了文本

样本聚合器：

序列大小为 2 时，当我手动迭代使用 msgGroup 分组的每条消息的时间戳时，仍有一些（不是全部）消息的差异超过 1000 毫秒。有什么我错过的吗？

请注意，相关 ID 和序列号是根据有效负载以编程方式设置的。因此，某些消息可能具有与序列号相同的相关 ID。