我想创建一个聚合器，用来统计来自源（例如SFTP服务器）的文件数量，并将其发送到一个度量工具。当存在文件时，结果应该发送为0。

我提出了以下解决方案，当有触发聚合的交换存在时它工作正常，但没有触发聚合时却什么也没发生。

from(getSourceUri())
        .onException(Throwable.class).bean(exceptionHandler).handled(true).end()
        .process(sftpProcessor)
        .aggregate(constant(true), new OverviewAggregatorStrategy())
            .aggregationStrategyMethodAllowNull()
            .completionInterval(TWO_MINUTES)
        .process(overviewProcessor)
        .end();

我使用Spring Boot，OverviewAggregatorStrategy 和 overviewProcessor 非常简单：

public class OverviewAggregatorStrategy implements AggregationStrategy {
  @Override
  public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    Long processedMessages;
    if (oldExchange != null) {
      processedMessages = oldExchange.getProperty(PROCESSED_MESSAGES, Long.class);
      oldExchange.setProperty(PROCESSED_MESSAGES, processedMessages + 1);
      return oldExchange;
    } else {
      processedMessages = 0L;
      newExchange.setProperty(PROCESSED_MESSAGES, processedMessages + 1);
      return newExchange;
    }
  }
}

@Component
public class OverviewProcessor implements Processor {

  @Override
  public void process(Exchange exchange) throws Exception {
    Object processedMessages = exchange.getProperty(PROCESSED_MESSAGES);
    log.info("Processed {} in last 10 minutes...", processedMessages);
  }
}

我认为有两个一般性的问题：

1. 如果这是一个可行的解决方案，如何修复以实现如前所述的工作方式？；或
2. 对于这类工作，还有更好的做法吗？

解决方案

你可以在消费端启用此选项

启用此选项后，sftp消费者将创建一个空主体的交换，即使源sftp文件夹中没有要消费的文件。

然后在你的聚合策略中，你可以相应地处理带有空主体的交换，将它们视为数据缺失。

例如，如果主体为空，你就不需要对计数器进行+1的操作。