如何将 CircuitBreaker 与 TimeLimiter 和 Bulkhead 结合使用？

Posted 2023-02-27

【中文标题】如何将 CircuitBreaker 与 TimeLimiter 和 Bulkhead 结合使用？

【英文标题】：How to combine CircuitBreaker with TimeLimiter and Bulkhead?

【发布时间】：2020-09-23 21:26:02

【问题描述】：

我有一个通过 REST 调用依赖项的服务。服务和依赖是微服务架构的一部分，所以我想使用弹性模式。我的目标是：

有一个断路器以在依赖关系陷入困境时保护它 限制呼叫可以运行的时间。该服务具有 SLA，并且必须在特定时间内回答。超时时，我们使用回退值。 限制对依赖项的并发调用数。通常调用率很低，响应速度很快，但我们希望保护依赖关系免受服务内的突发和排队请求。

以下是我当前的代码。它可以工作，但理想情况下我想使用 TimeLimiter 和 Bulkhead 类，因为它们似乎可以一起工作。

我怎样才能写得更好？

@Component
class FooService(@Autowired val circuitBreakerRegistry: CircuitBreakerRegistry)

    ...

    // State machine to take load off the dependency when slow or unresponsive
    private val circuitBreaker = circuitBreakerRegistry
        .circuitBreaker("fooService")

    // Limit parallel requests to dependency
    private var semaphore = Semaphore(maxParallelRequests)

    // The protected function
    private suspend fun makeHttpCall(customerId: String): Boolean 
        val client = webClientProvider.getCachedWebClient(baseUrl)

        val response = client
            .head()
            .uri("/the/request/url")
            .awaitExchange()

        return when (val status = response.rawStatusCode()) 
            200 -> true
            204 -> false
            else -> throw Exception(
                "Foo service responded with invalid status code: $status"
            )
        
    

    // Main function
    suspend fun isFoo(someId: String): Boolean 
        try 
            return circuitBreaker.executeSuspendFunction 
                semaphore.withPermit 
                    try 
                        withTimeout(timeoutMs) 
                            makeHttpCall(someId)
                        
                     catch (e: TimeoutCancellationException) 
                        // This exception has to be converted because
                        // the circuit-breaker ignores CancellationException
                        throw Exception("Call to foo service timed out")
                    
                
            
         catch (e: CallNotPermittedException) 
            logger.error  "Call to foo blocked by circuit breaker" 
         catch (e: Exception) 
            logger.error  "Exception while calling foo service: $e.message" 
        

        // Fallback
        return true
    

理想情况下，我想为 Flows 写类似 docs 的描述：

// Main function
suspend fun isFoo(someId: String): Boolean 
    return monoOf(makeHttpCall(someId))
        .bulkhead(bulkhead)
        .timeLimiter(timeLimiter)
        .circuitBreaker(circuitBreaker)

【参考方案1】：

您也可以使用 Resilience4j 的 Bulkhead 代替您自己的 Semaphore 和 Resilience4j 的 TimeLimiter。 您可以将 CircuitBreaker 与 bulkhead.executeSuspendFunction 和 timelimiter.executeSuspendFunction 叠加。

【讨论】：

您好，感谢您的回复。我不确定 Bulkhead 是否会执行，这里有关于性能的警告：resilience4j.readme.io/docs/getting-started-4#bulkhead 使用TimeLimiter，我还需要转换异常吗？理想情况下，我不想嵌套这些调用，而是使用装饰器之类的东西。