java - 如何使用 Spring Integration Aggregator 进行 'look-ahead' 发布？

Question

问题

我正在使用 Spring Integration(版本 5.0.0.M2，因为我需要动态流注册功能)开发一个应用程序，并且在某些时候我必须根据消息的 correlationId 聚合消息 header 。 header 标记了消息中的一些连续的子序列，即，它在新的子序列开始时发生变化。聚合器的目的是将子序列转换为单独的“分组”消息。
消息来源是外部的且不可预测，因此它无法为发出的消息提供 sequenceSize header 。因此，只有在下一条消息带有另一个 correlationId header (或者在 timeout 毫秒内没有下一条消息)时，才能决定是否释放当前的累积组。表示输入结束)。部分释放是 Not Acceptable 。

问题是 Spring Integration ReleaseStrategy 被设计为仅释放当前累积组(包括当前消息)，而我的任务需要““向前看”下一条消息，以确定当前组是否完整。从下一条消息的角度来看，同样的问题可能会被认为是“回望”发布。

所以问题是:有没有办法根据下一条消息的 header (不包括后者)从聚合器中释放消息组？

尝试解决方案

我研究了org.springframework.integration.aggregator.AbstractCorrelatedMessageHandler的源代码，目前我发现的唯一解决方法是一个相当肮脏的技巧 - 我继承自AggregatingMessageHandler 并重写 handleMessageInternal 方法。在该方法中，我将当前消息的correlationId 与上一次调用时保存的correlationId 进行比较。如果不同，我从存储中提取上一个组并用它调用 forceComplete 。然后(以任何方式)我将当前消息处理委托(delegate)给父级的 handleMessageInternal 方法。这是代码片段:

  @Override
  protected void handleMessageInternal(Message<?> message) throws Exception {
    Long currentGroupId = message.getHeaders().get(CORRELATION_ID, Long.class);

    boolean needToReleasePreviousGroup = ((previousGroupId != null) && !previousGroupId.equals(currentGroupId));
    if (needToReleasePreviousGroup) {
      MessageGroup previousGroup = getMessageStore().getMessageGroup(previousGroupId);
      if (previousGroup.size() != 0) {
        forceComplete(previousGroup);
      } else {
        log.debug("Previous group with id={} has been already released. Skip.", previousGroupId);
      }
    }

    super.handleMessageInternal(message);
    previousGroupId = currentGroupId;   // do unconditionally as we'll check that group on the next step anyway
  }

我意识到这个解决方案使聚合器本身具有状态(虽然不应该如此)并且不适用于许多其他场景。此外，它使应用代码与框架的内部耦合。

如果有人向我指出更好的解决方案，我会很高兴。我愿意提供有关该问题或我所使用的解决方案的更多详细信息。

Answer 1

查看您的 previousGroupId 解决方案和描述:

'look ahead' for a next message to find out if current group is complete or not.

“当前”这个词是这里的关键。

因此，对我来说，这意味着您的聚合器是单个组，您必须释放当前状态并根据当前消息的状态启动一个新状态。

为此目的，我将创建一个真正的单分组聚合器 - correlationStrategy 应该返回一些常量，例如1。

是否释放当前组实际上是 ReleaseStrategy 的责任。最新消息实际上是 MessageGroup.getMessages() 中的最后一条消息。因此，您到达该消息，决定是否需要释放，然后继续使用自定义 MessageGroupProcessor 以从聚合器生成累积结果。

在 MessageGroupProcessor 中，您删除最新的“有罪”消息，不要将其包含到结果中，并将其发送回聚合器以形成新的消息组。

这对你来说有意义吗？