Блог

У нас есть процесс разделения, который отправляет сообщения в разные очереди. Существует другой процесс, который собирает и объединяет эти сообщения для дальнейшей обработки.

Мы хотим иметь тайм-аут между моментом разделения и агрегированием. Тайм-аут агрегации IIUC начинается с первого сообщения и сбрасывается после каждого агрегированного сообщения (он основан на интервале, а не для полного сообщения).

Какое лучшее решение для решения этой проблемы?

Редактировать

Вот лучшее, что я смог придумать, хотя это немного взлом. Сначала вы сохраняете временную метку в качестве заголовка сообщения и публикуете ее в очереди вместе с телом:

 from("somewhere")
    .split(body())
    .process(e -> e.getIn().setHeader("aggregation_timeout", 
        ZonedDateTime.now().plusSeconds(COMPLETION_TIMEOUT)))
    .to("aggregation-route-uri");
  

Затем, при потреблении и агрегировании, вы используете пользовательскую стратегию агрегирования, которая сохранит aggregation_timeout из первого сообщения в текущей группе, а затем использует a, completionPredicate который считывает это значение, чтобы проверить, истек ли тайм-аут (альтернативно, если вы агрегируете таким образом, чтобы порядок сообщений оставался неизменным, вы могли бы простопрочитайте заголовок из первого сообщения). Используйте short completionTimeout в качестве меры предосторожности для случаев, когда интервал между двумя сообщениями большой:

 from("aggregation-route-uri")
    .aggregate(bySomething())
    .aggregationStrategy((oldExchange, newExchange) -> {
      // read aggregation_timeout header from first message 
      // and set it as property in grouped exchange
      // perform aggregation
    })
    .completionTimeout(1000) // intentionally low value, here as a safeguard
    .completionPredicate(e -> {
      // complete once the timeout has been reached
      return e.getProperty("aggregation_timeout", ZonedDateTime.class)
              .isAfter(ZonedDateTime.now());
    })
    .process(e -> // do something with aggregates);