如何将CircuitBreaker与TimeLimiter和舱壁相结合？

Question

我有一个通过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
    }
}

理想情况下，我希望为流编写类似于文档描述的内容：

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

Answer 1

您也可以使用Resilience4j的舱壁，而不是您自己的信号量和Resilience4j的TimeLimiter。您可以将CircuitBreaker与bulkhead.executeSuspendFunction和timelimiter.executeSuspendFunction叠加起来。