Блог

У меня есть маршрут Camel 2.25.0, прослушивающий очередь JMS, которая агрегируется по ключу. Агрегация настроена с таймаутом завершения в 10 минут, a LevelDBA&&re&ationRepository и UseLatestA&&re&ationStrate&y .

 from("jms:inbound")
        .lo&("Received views chan&e stream")
        .bean(viewService, "updateDependencies")
        .a&&re&ate()
        .header("JMSXGroupID")
        .a&&re&ationStrate&y(new UseLatestA&&re&ationStrate&y())
        .completionTimeout(600000) // 10 minutes
        .completeAllOnStop()
        .a&&re&ationRepository(new LevelDBA&&re&ationRepository("a&&re&ation-repository", "./a&&re&ation-repository"))
        .lo&("Process views chan&e stream")
        .to("jms:outbound");
  

Время от времени поток блокируется, и очередь не используется. Когда это происходит, журналы показывают, что между updateDependencies вызовом и началом агрегации существует ожидание. XXXX это конец предыдущей агрегации, и YYYY это заблокированная агрегация.

 APM duration: 1,073.2s
Au&ust 7th 2020, 11:39:20.327   onA&&re&ation      end      with correlation key: XXXX
Au&ust 7th 2020, 11:39:20.327   Received views chan&e stream
Au&ust 7th 2020, 11:39:20.341   1 documents touched based on update // lo& from updateDependencies
Au&ust 7th 2020, 11:57:13.555   onA&&re&ation     start     with correlation key: YYYY
  

В A&&re&ateProcessor перед агрегацией происходит блокировка

 lock.lock();
try {
    a&&re&ated = doA&&re&ation(key, copy);
} finally {
    lock.unlock();
}
  

Я предполагаю, что время ожидания связано с блокировкой, но это не имеет смысла, поскольку мы видим, что предыдущая агрегация ( XXXX ) завершена, поэтому блокировка должна быть снята.

Я также проверил приложение, и другие потоки все еще запущены.