Spring Integration Flow: автоматический выключатель для каждой конечной точки или на уровне потока

Question

Я успешно реализовал некоторый пружинный поток интеграции.

Я хочу, чтобы автоматический выключатель был либо одинаковым для каждой конечной точки, либо на уровне потока. Я уже прочитал эту документацию https://docs.spring.io/spring-integration/reference/html/handler-advice.html, но я не нашел свой ответ.

Должен ли я использовать AOP?

Спасибо

G.

Answer 1

Я не уверен, что вы пропустили в упомянутых документах, но RequestHandlerCircuitBreakerAdvice действительно там: https://docs.spring.io/spring-integration/reference/html/handler-advice.html#circuit -breaker-advice

Такие советы должны быть примененным в Java DSL с этой опцией конфигурации:

.transform(..., c -> c.advice(expressionAdvice()))

Обратите внимание на этот вызов advice(expressionAdvice()). expressionAdvice() - это метод бобов. Таким образом, вы можете сделать нечто подобное для RequestHandlerCircuitBreakerAdvice и любых ваших конечных точек в потоке, которые должны быть защищены схемой.

И да, вы можете использовать только один компонент для RequestHandlerCircuitBreakerAdvice. Он сохраняет состояние для любой конечной точки, против которой он вызывается:

protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
    AdvisedMetadata metadata = this.metadataMap.get(target);
    if (metadata == null) {
        this.metadataMap.putIfAbsent(target, new AdvisedMetadata());
        metadata = this.metadataMap.get(target);
    }

Answer 2

Спасибо за ваш ответ @ артем-билан. Я действительно ценю, что член команды весенней интеграции ответил на это.

После долгих размышлений я переформулировал свою проблему.

Учитывая IntegrationFlow, с указанным c каналом ошибки, если есть больше заданного количества ошибок за заданный промежуток времени (более 10 ошибок за 10 с), я хочу прекратить опрашивать входной канал.

Поэтому я перенаправляю все ошибки для этого потока на указанный c канал ошибки потока. Счетчик ошибок увеличивается, и затем, если пороговое значение достигается за заданный промежуток времени, я останавливаю опросчик.

У меня есть второй поток, который отслеживает "остановленные" обработчики запросов и через некоторое время перезапускает их.