摘要：以下是它的示例現(xiàn)在，如果需要的話，所有對(duì)已修飾的服務(wù)塊的調(diào)用都要符合速率限制器配置。

案例概述

在本文中，我們討論一下Resilience4j庫。
該庫通過管理遠(yuǎn)程通信的容錯(cuò)性來幫助實(shí)現(xiàn)彈性系統(tǒng)。
這個(gè)庫受到Hystrix的啟發(fā)，但提供了更方便的API和許多其他特性，如速率限制器(阻塞太頻繁的請(qǐng)求)、Bulkhead(避免太多并發(fā)請(qǐng)求)等。

Maven設(shè)置

首先，我們需要將目標(biāo)模塊添加到我們的pom.xml中（例如，我們添加了Circuit Breaker）：

????io.github.resilience4j
????resilience4j-circuitbreaker
????0.12.1

在這里，我們使用的是斷路器模塊。所有模塊及其最新版本均可在Maven Central上找到。
在接下來的部分中，我們將介紹庫中最常用的模塊。

斷路器

請(qǐng)注意，對(duì)于此模塊，我們需要上面顯示的設(shè)置resilience4j-circuitbreaker依賴項(xiàng)。

斷路器模式可以幫助我們?cè)谶h(yuǎn)程服務(wù)中斷時(shí)防止一連串的故障。

在多次失敗的嘗試之后，我們可以認(rèn)為服務(wù)不可用/重載，并急切地拒絕所有后續(xù)的請(qǐng)求。通過這種方式，我們可以為可能失敗的調(diào)用節(jié)省系統(tǒng)資源。

讓我們看看我們?nèi)绾瓮ㄟ^Resilience4j實(shí)現(xiàn)這一目標(biāo)。

首先，我們需要定義要使用的設(shè)置。最簡(jiǎn)單的方法是使用默認(rèn)設(shè)置：

CircuitBreakerRegistry circuitBreakerRegistry
??= CircuitBreakerRegistry.ofDefaults();

也可以使用自定義參數(shù)：

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
??.failureRateThreshold(20)
??.ringBufferSizeInClosedState(5)
??.build();

在這里，我們將速率閾值設(shè)置為20％，并將嘗試呼叫設(shè)置為最少5次。

然后，我們創(chuàng)建一個(gè)CircuitBreaker對(duì)象并通過它調(diào)用遠(yuǎn)程服務(wù)：

interface RemoteService {
????int process(int i);
}
?
CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(config);
CircuitBreaker circuitBreaker = registry.circuitBreaker("my");
Function decorated = CircuitBreaker
??.decorateFunction(circuitBreaker, service::process);

最后，讓我們通過JUnit測(cè)試看看它是如何工作的。

我們將嘗試調(diào)用服務(wù)10次。我們應(yīng)該能夠驗(yàn)證該呼叫至少嘗試了5次，然后在20%的呼叫失敗后停止:

when(service.process(any(Integer.class))).thenThrow(new RuntimeException());
?
for (int i = 0; i < 10; i++) {
????try {
????????decorated.apply(i);
????} catch (Exception ignore) {}
}
?
verify(service, times(5)).process(any(Integer.class));

斷路器的狀態(tài)和設(shè)置

斷路器可以處于以下三種狀態(tài)之一:

CLOSED - 一切正常，不涉及短路

OPEN - 遠(yuǎn)程服務(wù)器已關(guān)閉，所有請(qǐng)求都被短路

HALF_OPEN - 從進(jìn)入開放狀態(tài)到現(xiàn)在已經(jīng)經(jīng)過了一段時(shí)間，斷路器允許請(qǐng)求檢查遠(yuǎn)程服務(wù)是否重新上線

我們可以配置以下設(shè)置:

斷路器打開并開始短路呼叫的故障率閾值

等待時(shí)間，它定義了斷路器在切換到半開狀態(tài)之前應(yīng)該保持打開狀態(tài)的時(shí)間

斷路器半開或半閉時(shí)環(huán)形緩沖器的尺寸

處理斷路器事件的定制電路斷路器事件監(jiān)聽器

一個(gè)自定義謂詞，用于評(píng)估異常是否應(yīng)算作故障，從而提高故障率

速率限制器

與上一節(jié)類似，此功能需要resilience4j-ratelimiter依賴項(xiàng)。

顧名思義，此功能允許限制對(duì)某些服務(wù)的訪問。它的API與CircuitBreaker非常相似- 有Registry，Config和Limiter類。

以下是它的示例：

RateLimiterConfig config = RateLimiterConfig.custom().limitForPeriod(2).build();
RateLimiterRegistry registry = RateLimiterRegistry.of(config);
RateLimiter rateLimiter = registry.rateLimiter("my");
Function decorated
??= RateLimiter.decorateFunction(rateLimiter, service::process);

現(xiàn)在，如果需要的話，所有對(duì)已修飾的服務(wù)塊的調(diào)用都要符合速率限制器配置。

我們可以配置如下參數(shù)：

限制刷新的時(shí)間段

刷新周期的權(quán)限限制

默認(rèn)等待權(quán)限持續(xù)時(shí)間

Bulkhead

在這里，我們首先需要relasticience4j-bulkhead依賴項(xiàng)。

可以限制對(duì)特定服務(wù)的并發(fā)調(diào)用數(shù)。

讓我們看一個(gè)使用Bulkhead API配置最多一個(gè)并發(fā)調(diào)用的示例：

BulkheadConfig config = BulkheadConfig.custom().maxConcurrentCalls(1).build();
BulkheadRegistry registry = BulkheadRegistry.of(config);
Bulkhead bulkhead = registry.bulkhead("my");
Function decorated
??= Bulkhead.decorateFunction(bulkhead, service::process);

要測(cè)試此配置，我們將調(diào)用模擬服務(wù)的方法。

然后，我們確保Bulkhead不允許任何其他調(diào)用：

CountDownLatch latch = new CountDownLatch(1);
when(service.process(anyInt())).thenAnswer(invocation -> {
????latch.countDown();
????Thread.currentThread().join();
????return null;
});
?
ForkJoinTask task = ForkJoinPool.commonPool().submit(() -> {
????try {
????????decorated.apply(1);
????} finally {
????????bulkhead.onComplete();
????}
});
latch.await();
assertThat(bulkhead.isCallPermitted()).isFalse();

我們可以配置以下設(shè)置：

Bulkhead允許的最大并行執(zhí)行量

嘗試進(jìn)入飽和艙壁時(shí)線程等待的最長時(shí)間

重試

對(duì)于此功能，我們需要將resilience4j-retry庫添加到項(xiàng)目中。

我們可以使用Retry API 自動(dòng)重試失敗的呼叫：

RetryConfig config = RetryConfig.custom().maxAttempts(2).build();
RetryRegistry registry = RetryRegistry.of(config);
Retry retry = registry.retry("my");
Function decorated
??= Retry.decorateFunction(retry, (Integer s) -> {
????????service.process(s);
????????return null;
????});

現(xiàn)在讓我們模擬在遠(yuǎn)程服務(wù)調(diào)用期間拋出異常的情況，并確保庫自動(dòng)重試失敗的調(diào)用：

when(service.process(anyInt())).thenThrow(new RuntimeException());
try {
????decorated.apply(1);
????fail("Expected an exception to be thrown if all retries failed");
} catch (Exception e) {
????verify(service, times(2)).process(any(Integer.class));
}

我們還可以配置以下內(nèi)容：

最大嘗試次數(shù)

重試前的等待時(shí)間

自定義函數(shù)，用于修改失敗后的等待間隔

自定義謂詞，用于評(píng)估異常是否應(yīng)導(dǎo)致重試調(diào)用

緩存

Cache模塊需要resilience4j-cache依賴項(xiàng)。

初始化看起來與其他模塊略有不同：

javax.cache.Cache cache = ...; // Use appropriate cache here
Cache cacheContext = Cache.of(cache);
Function decorated
??= Cache.decorateSupplier(cacheContext, () -> service.process(1));

這里的緩存是通過使用JSR-107 Cache實(shí)現(xiàn)完成的，Resilience4j提供了一種應(yīng)用它的方法。

請(qǐng)注意，沒有用于裝飾功能的API（如Cache.decorateFunction（Function）），API僅支持 Supplier和Callable類型。

TimeLimiter

對(duì)于此模塊，我們必須添加resilience4j-timelimiter依賴項(xiàng)。
可以使用TimeLimiter限制調(diào)用遠(yuǎn)程服務(wù)所花費(fèi)的時(shí)間。
為了演示，讓我們?cè)O(shè)置一個(gè)配置超時(shí)為1毫秒的TimeLimiter：

long ttl = 1;
TimeLimiterConfig config
??= TimeLimiterConfig.custom().timeoutDuration(Duration.ofMillis(ttl)).build();
TimeLimiter timeLimiter = TimeLimiter.of(config);

接下來，讓我們驗(yàn)證Resilience4j是否使用預(yù)期的超時(shí)調(diào)用Future.get()：

Future futureMock = mock(Future.class);
Callable restrictedCall
??= TimeLimiter.decorateFutureSupplier(timeLimiter, () -> futureMock);
restrictedCall.call();
?
verify(futureMock).get(ttl, TimeUnit.MILLISECONDS);

我們也可以將它與CircuitBreaker結(jié)合使用：

Callable chainedCallable
??= CircuitBreaker.decorateCallable(circuitBreaker, restrictedCall);

附加模塊

Resilience4j還提供了許多附加模塊，可以簡(jiǎn)化與流行框架和庫的集成。

一些比較知名的集成是：

Spring Boot – resilience4j-spring-boot

Ratpack – resilience4j-ratpack

Retrofit – resilience4j-retrofit

Vertx – resilience4j-vertx

Dropwizard – resilience4j-metrics

Prometheus – resilience4j-prometheus

案例結(jié)論

在本文中，我們了解了Resilience4j庫的各個(gè)方面，并學(xué)習(xí)了如何使用它來解決服務(wù)器間通信中的各種容錯(cuò)問題。