异步微服务的断路器..？

Question

有一个 ActiveMQ 队列 (QueueA)。服务 (MyService) 订阅消息、处理消息并将消息发送到另一个 ActiveMQ 队列 (QueueB)。

QueueA -> MyService -> QueueB

假设有数千条消息位于QueueA 中。与此同时，QueueB 已下线。如果在向QueueB 发送消息时有一定数量的消息（比如 100 条）消息连续失败，我想停止处理。它应该在特定时间段内测试滚动窗口（例如，100 条连续消息在 60 秒内失败）并停止从 QueueA 消费。然后，它应该通过再发送一条消息来测试服务是否在 15 分钟左右后启动。如果仍然失败，请再次停止从 QueueA 消费 15 分钟。

现在，所有消息都出错了，我们必须再次重新处理每条消息。有一个恢复机制，但是由于当前架构的限制，恢复机制变得过载。

这有什么模式吗？它是同一个断路器吗（我在同步上下文中知道它）。如果是这样，不确定 Java / Spring Boot / Apache Camel 中是否有解决方案。是的，这就是我们目前使用的技术堆栈。即使您可能不了解此特定技术平台，该模式的任何指南也会有所帮助。

我还在 StackOverflow 中阅读了以下问题。

Is circuit breaker pattern applicable for asynchronous requests also?

感谢您抽出宝贵时间帮助我。

Answer 1

查看基于 CircuitBreakerLoadBalancer 的 ThrottlingExceptionRoutePolicy 类型的 Camel RoutePolicy。

使用此策略应该允许您在电路处于打开状态时停止从端点消费（与标准电路行为比较：绕过服务调用，并回退到另一个响应）。

@Bean
public ThrottlingExceptionRoutePolicy myCustomPolicy() {
    // Important: do not open circuit for this kind of exceptions
    List<Class<?>> handledExceptions = Arrays.asList(MyException.class);
    return new ThrottlingExceptionRoutePolicy(failureThreshold, failureWindow, halfOpenAfter, handledExceptions);
} 
 
from("jms:queue:QueueA")
    .routePolicy(myCustomPolicy)
    .to("mock:MyService")