當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Resilience4j-轻量级熔断框架

發布時間：2025/3/21 编程问答 32 豆豆

生活随笔收集整理的這篇文章主要介紹了 Resilience4j-轻量级熔断框架小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Resilience4j

簡介

Resilience4j是一款輕量級，易于使用的容錯庫，其靈感來自于Netflix Hystrix，但是專為Java 8和函數式編程而設計。輕量級，因為庫只使用了Vavr，它沒有任何其他外部依賴下。相比之下，Netflix Hystrix對Archaius具有編譯依賴性，Archaius具有更多的外部庫依賴性，例如Guava和Apache Commons Configuration。

要使用Resilience4j，不需要引入所有依賴，只需要選擇你需要的。

Resilience4j提供了以下的核心模塊和拓展模塊:

核心模塊：

resilience4j-circuitbreaker: Circuit breaking
resilience4j-ratelimiter: Rate limiting
resilience4j-bulkhead: Bulkheading
resilience4j-retry: Automatic retrying (sync and async)
resilience4j-cache: Result caching
resilience4j-timelimiter: Timeout handling

Circuitbreaker

簡介

CircuitBreaker通過具有三種正常狀態的有限狀態機實現：CLOSED，OPEN和HALF_OPEN以及兩個特殊狀態DISABLED和FORCED_OPEN。當熔斷器關閉時，所有的請求都會通過熔斷器。如果失敗率超過設定的閾值，熔斷器就會從關閉狀態轉換到打開狀態，這時所有的請求都會被拒絕。當經過一段時間后，熔斷器會從打開狀態轉換到半開狀態，這時僅有一定數量的請求會被放入，并重新計算失敗率，如果失敗率超過閾值，則變為打開狀態，如果失敗率低于閾值，則變為關閉狀態。

Circuitbreaker狀態機

Resilience4j記錄請求狀態的數據結構和Hystrix不同，Hystrix是使用滑動窗口來進行存儲的，而Resilience4j采用的是Ring Bit Buffer(環形緩沖區)。Ring Bit Buffer在內部使用BitSet這樣的數據結構來進行存儲，BitSet的結構如下圖所示：

環形緩沖區

每一次請求的成功或失敗狀態只占用一個bit位，與boolean數組相比更節省內存。BitSet使用long[]數組來存儲這些數據，意味著16個值(64bit)的數組可以存儲1024個調用狀態。

計算失敗率需要填滿環形緩沖區。例如，如果環形緩沖區的大小為10，則必須至少請求滿10次，才會進行故障率的計算，如果僅僅請求了9次，即使9個請求都失敗，熔斷器也不會打開。但是CLOSE狀態下的緩沖區大小設置為10并不意味著只會進入10個請求，在熔斷器打開之前的所有請求都會被放入。

當故障率高于設定的閾值時，熔斷器狀態會從由CLOSE變為OPEN。這時所有的請求都會拋出CallNotPermittedException異常。當經過一段時間后，熔斷器的狀態會從OPEN變為HALF_OPEN，HALF_OPEN狀態下同樣會有一個Ring Bit Buffer，用來計算HALF_OPEN狀態下的故障率，如果高于配置的閾值，會轉換為OPEN，低于閾值則裝換為CLOSE。與CLOSE狀態下的緩沖區不同的地方在于，HALF_OPEN狀態下的緩沖區大小會限制請求數，只有緩沖區大小的請求數會被放入。

除此以外，熔斷器還會有兩種特殊狀態：DISABLED（始終允許訪問）和FORCED_OPEN（始終拒絕訪問）。這兩個狀態不會生成熔斷器事件（除狀態裝換外），并且不會記錄事件的成功或者失敗。退出這兩個狀態的唯一方法是觸發狀態轉換或者重置熔斷器。

熔斷器關于線程安全的保證措施有以下幾個部分：

熔斷器的狀態使用AtomicReference保存的
更新熔斷器狀態是通過無狀態的函數或者原子操作進行的
更新事件的狀態用synchronized關鍵字保護

意味著同一時間只有一個線程能夠修改熔斷器狀態或者記錄事件的狀態。

可配置參數

配置參數默認值描述

failureRateThreshold	50	熔斷器關閉狀態和半開狀態使用的同一個失敗率閾值
ringBufferSizeInHalfOpenState	10	熔斷器半開狀態的緩沖區大小，會限制線程的并發量，例如緩沖區為10則每次只會允許10個請求調用后端服務
ringBufferSizeInClosedState	100	熔斷器關閉狀態的緩沖區大小，不會限制線程的并發量，在熔斷器發生狀態轉換前所有請求都會調用后端服務
waitDurationInOpenState	60(s)	熔斷器從打開狀態轉變為半開狀態等待的時間
automaticTransitionFromOpenToHalfOpenEnabled	false	如果置為true，當等待時間結束會自動由打開變為半開，若置為false，則需要一個請求進入來觸發熔斷器狀態轉換
recordExceptions	empty	需要記錄為失敗的異常列表
ignoreExceptions	empty	需要忽略的異常列表
recordFailure	throwable -> true	自定義的謂詞邏輯用于判斷異常是否需要記錄或者需要忽略，默認所有異常都進行記錄

測試前準備

pom.xml

測試使用的IDE為idea，使用的springboot進行學習測試，首先引入maven依賴：

<dependency><groupId>io.github.resilience4j</groupId><artifactId>resilience4j-spring-boot</artifactId><version>0.9.0</version> </dependency>

resilience4j-spring-boot集成了circuitbeaker、retry、bulkhead、ratelimiter幾個模塊，因為后續還要學習其他模塊，就直接引入resilience4j-spring-boot依賴。

application.yml配置

resilience4j:circuitbreaker:configs:default:ringBufferSizeInClosedState: 5 # 熔斷器關閉時的緩沖區大小ringBufferSizeInHalfOpenState: 2 # 熔斷器半開時的緩沖區大小waitDurationInOpenState: 10000 # 熔斷器從打開到半開需要的時間failureRateThreshold: 60 # 熔斷器打開的失敗閾值eventConsumerBufferSize: 10 # 事件緩沖區大小registerHealthIndicator: true # 健康監測automaticTransitionFromOpenToHalfOpenEnabled: false # 是否自動從打開到半開，不需要觸發recordFailurePredicate: com.example.resilience4j.exceptions.RecordFailurePredicate # 謂詞設置異常是否為失敗recordExceptions: # 記錄的異常- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAExceptionignoreExceptions: # 忽略的異常- com.example.resilience4j.exceptions.BusinessAExceptioninstances:backendA:baseConfig: defaultwaitDurationInOpenState: 5000failureRateThreshold: 20backendB:baseConfig: default

可以配置多個熔斷器實例，使用不同配置或者覆蓋配置。

需要保護的后端服務

以一個查找用戶列表的后端服務為例，利用熔斷器保護該服務。

interface RemoteService {List<User> process() throws TimeoutException, InterruptedException; }

連接器調用該服務

這是調用遠端服務的連接器，我們通過調用連接器中的方法來調用后端服務。

public RemoteServiceConnector{public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;} }

用于監控熔斷器狀態及事件的工具類

要想學習各個配置項的作用，需要獲取特定時候的熔斷器狀態，寫一個工具類：

@Log4j2 public class CircuitBreakerUtil {/*** @Description: 獲取熔斷器的狀態*/public static void getCircuitBreakerStatus(String time, CircuitBreaker circuitBreaker){CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();// Returns the failure rate in percentage.float failureRate = metrics.getFailureRate();// Returns the current number of buffered calls.int bufferedCalls = metrics.getNumberOfBufferedCalls();// Returns the current number of failed calls.int failedCalls = metrics.getNumberOfFailedCalls();// Returns the current number of successed calls.int successCalls = metrics.getNumberOfSuccessfulCalls();// Returns the max number of buffered calls.int maxBufferCalls = metrics.getMaxNumberOfBufferedCalls();// Returns the current number of not permitted calls.long notPermittedCalls = metrics.getNumberOfNotPermittedCalls();log.info(time + "state=" +circuitBreaker.getState() + " , metrics[ failureRate=" + failureRate +", bufferedCalls=" + bufferedCalls +", failedCalls=" + failedCalls +", successCalls=" + successCalls +", maxBufferCalls=" + maxBufferCalls +", notPermittedCalls=" + notPermittedCalls +" ]");}/*** @Description: 監聽熔斷器事件*/public static void addCircuitBreakerListener(CircuitBreaker circuitBreaker){circuitBreaker.getEventPublisher().onSuccess(event -> log.info("服務調用成功：" + event.toString())).onError(event -> log.info("服務調用失敗：" + event.toString())).onIgnoredError(event -> log.info("服務調用失敗，但異常被忽略：" + event.toString())).onReset(event -> log.info("熔斷器重置：" + event.toString())).onStateTransition(event -> log.info("熔斷器狀態改變：" + event.toString())).onCallNotPermitted(event -> log.info(" 熔斷器已經打開：" + event.toString()));}

調用方法

CircuitBreaker目前支持兩種方式調用，一種是程序式調用，一種是AOP使用注解的方式調用。

程序式的調用方法

在CircuitService中先注入注冊器，然后用注冊器通過熔斷器名稱獲取熔斷器。如果不需要使用降級函數，可以直接調用熔斷器的executeSupplier方法或executeCheckedSupplier方法：

public class CircuitBreakerServiceImpl{@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerNotAOP() throws Throwable {CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執行開始前：", circuitBreaker);circuitBreaker.executeCheckedSupplier(remotServiceConnector::process);} }

如果需要使用降級函數，則要使用decorate包裝服務的方法，再使用Try.of().recover()進行降級處理，同時也可以根據不同的異常使用不同的降級方法：

public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerNotAOP(){// 通過注冊器獲取熔斷器的實例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執行開始前：", circuitBreaker);// 使用熔斷器包裝連接器的方法CheckedFunction0<List<User>> checkedSupplier = CircuitBreaker.decorateCheckedSupplier(circuitBreaker, remoteServiceConnector::process);// 使用Try.of().recover()調用并進行降級處理Try<List<User>> result = Try.of(checkedSupplier).recover(CallNotPermittedException.class, throwable -> {log.info("熔斷器已經打開，拒絕訪問被保護方法~");CircuitBreakerUtil.getCircuitBreakerStatus("熔斷器打開中:", circuitBreaker);List<User> users = new ArrayList();return users;}).recover(throwable -> {log.info(throwable.getLocalizedMessage() + ",方法被降級了~~");CircuitBreakerUtil.getCircuitBreakerStatus("降級方法中:",circuitBreaker);List<User> users = new ArrayList();return users;});CircuitBreakerUtil.getCircuitBreakerStatus("執行結束后：", circuitBreaker);return result.get();} }

AOP式的調用方法

首先在連接器方法上使用@CircuitBreaker(name="",fallbackMethod="")注解，其中name是要使用的熔斷器的名稱，fallbackMethod是要使用的降級方法，降級方法必須和原方法放在同一個類中，且降級方法的返回值需要和原方法相同，輸入參數需要添加額外的exception參數，類似這樣：

public RemoteServiceConnector{@CircuitBreaker(name = "backendA", fallbackMethod = "fallBack")public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;}private List<User> fallBack(Throwable throwable){log.info(throwable.getLocalizedMessage() + ",方法被降級了~~");CircuitBreakerUtil.getCircuitBreakerStatus("降級方法中:", circuitBreakerRegistry.circuitBreaker("backendA"));List<User> users = new ArrayList();return users;}private List<User> fallBack(CallNotPermittedException e){log.info("熔斷器已經打開，拒絕訪問被保護方法~");CircuitBreakerUtil.getCircuitBreakerStatus("熔斷器打開中:", circuitBreakerRegistry.circuitBreaker("backendA"));List<User> users = new ArrayList();return users;}}

可使用多個降級方法，保持方法名相同，同時滿足的條件的降級方法會觸發最接近的一個（這里的接近是指類型的接近，先會觸發離它最近的子類異常），例如如果process()方法拋出CallNotPermittedException，將會觸發fallBack(CallNotPermittedException e)方法而不會觸發fallBack(Throwable throwable)方法。

之后直接調用方法就可以了：

public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerAOP() throws TimeoutException, InterruptedException {CircuitBreakerUtil.getCircuitBreakerStatus("執行開始前：",circuitBreakerRegistry.circuitBreaker("backendA"));List<User> result = remoteServiceConnector.process();CircuitBreakerUtil.getCircuitBreakerStatus("執行結束后：", circuitBreakerRegistry.circuitBreaker("backendA"));return result;} }

使用測試

接下來進入測試，首先我們定義了兩個異常，異常A同時在黑白名單中，異常B只在黑名單中：

然后對被保護的后端接口進行如下的實現：

public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() {int num = count.getAndIncrement();log.info("count的值 = " + num);if (num % 4 == 1){throw new BusinessAException("異常A，不需要被記錄");}if (num % 4 == 2 || num % 4 == 3){throw new BusinessBException("異常B，需要被記錄");}log.info("服務正常運行，獲取用戶列表");// 模擬數據庫的正常查詢return repository.findAll();} }

使用CircuitBreakerServiceImpl中的AOP或者程序式調用方法進行單元測試，循環調用10次：

public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerTest() {for (int i=0; i<10; i++){// circuitService.circuitBreakerAOP();circuitService.circuitBreakerNotAOP();}} }

看下運行結果：

執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 0 服務正常運行，獲取用戶列表執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 1 異常A，不需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 2 異常B，需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 3 異常B，需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 4 服務正常運行，獲取用戶列表執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 5 異常A，不需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 6 異常B，需要被記錄,方法被降級了~~ 降級方法中:state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器已經打開，拒絕訪問被保護方法~ 熔斷器打開中:state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ] 執行結束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ]

注意到異常A發生的前后bufferedCalls、failedCalls、successCalls三個參數的值都沒有沒有發生變化，說明白名單的優先級高于黑名單，源碼中也有提到Ignoring an exception has priority over recording an exception：

/** * @see #ignoreExceptions(Class[]) ). Ignoring an exception has priority over recording an exception. * <p> * Example: * recordExceptions(Throwable.class) and ignoreExceptions(RuntimeException.class) * would capture all Errors and checked Exceptions, and ignore unchecked * <p> */

同時也可以看出白名單所謂的忽略，是指不計入緩沖區中（即不算成功也不算失敗），有降級方法會調用降級方法，沒有降級方法會拋出異常，和其他異常無異。

執行開始前：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器已經打開，拒絕訪問被保護方法~ 熔斷器打開中:state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ] 執行結束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ]

當環形緩沖區大小被填滿時會計算失敗率，這時請求會被拒絕獲取不到count的值，且notPermittedCalls會增加。

接下來我們實驗一下多線程下熔斷器關閉和熔斷器半開兩種情況下緩沖環的區別，我們先開15個線程進行調用測試熔斷器關閉時的緩沖環，熔斷之后等10s再開15個線程進行調用測試熔斷器半開時的緩沖環：

public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerThreadTest() throws InterruptedException {ExecutorService pool = Executors.newCachedThreadPool();for (int i=0; i<15; i++){pool.submit(// circuitService::circuitBreakerAOPcircuitService::circuitBreakerNotAOP);}pool.shutdown();while (!pool.isTerminated());Thread.sleep(10000);log.info("熔斷器狀態已轉為半開");pool = Executors.newCachedThreadPool();for (int i=0; i<15; i++){pool.submit(// circuitService::circuitBreakerAOPcircuitService::circuitBreakerNotAOP);}pool.shutdown();while (!pool.isTerminated());for (int i=0; i<10; i++){}} }

前15個線程都通過了熔斷器，由于正常返回需要查數據庫，所以會慢很多，失敗率很快就達到了100%，而且觀察到如下的記錄：

異常B，需要被記錄,方法被降級了~~ 降級方法中:state=OPEN , metrics[ failureRate=100.0, bufferedCalls=5, failedCalls=5, successCalls=0, maxBufferCalls=5, notPermittedCalls=0 ]

可以看出，雖然熔斷器已經打開了，可是異常B還是進入了降級方法，拋出的異常不是notPermittedCalls數量為0，說明在熔斷器轉換成打開之前所有請求都通過了熔斷器，緩沖環不會控制線程的并發。

執行結束后：state=OPEN , metrics[ failureRate=80.0, bufferedCalls=5, failedCalls=4, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=OPEN , metrics[ failureRate=20.0, bufferedCalls=5, failedCalls=1, successCalls=4, maxBufferCalls=5, notPermittedCalls=0 ]

同時以上幾條正常執行的服務完成后，熔斷器的失敗率在下降，說明熔斷器打開狀態下還是會計算失敗率，由于環形緩沖區大小為5，初步推斷成功的狀態會依次覆蓋最開始的幾個狀態，所以得到了上述結果。

接下來分析后15個線程的結果

熔斷器狀態已轉為半開執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 16 服務正常運行，獲取用戶列表執行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器狀態改變：2019-07-29T17:19:19.959+08:00[Asia/Shanghai]: CircuitBreaker 'backendA' changed state from OPEN to HALF_OPEN count的值 = 18 count的值 = 17 服務正常運行，獲取用戶列表 count的值 = 19 count的值 = 15

熔斷器狀態從打開到半開我設置的是5s，前15個線程調用之后我等待了10s，熔斷器應該已經變為半開了，但是執行開始前熔斷器的狀態卻是OPEN，這是因為默認的配置項automaticTransitionFromOpenToHalfOpenEnabled=false，時間到了也不會自動轉換，需要有新的請求來觸發熔斷器的狀態轉換。同時我們發現，好像狀態改變后還是進了超過4個請求，似乎半開狀態的環并不能限制線程數？這是由于這些進程是在熔斷器打開時一起進來的。為了更好的觀察環半開時候環大小是否限制線程數，我們修改一下配置：

resilience4j:circuitbreaker:configs:myDefault:automaticTransitionFromOpenToHalfOpenEnabled: true # 是否自動從打開到半開

我們再試一次：

熔斷器狀態已轉為半開執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] count的值 = 15 count的值 = 16 服務正常運行，獲取用戶列表異常B，需要被記錄,方法被降級了~~ 降級方法中:state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=1, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行結束后：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=1, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] count的值 = 17 異常A，不需要被記錄,方法被降級了~~ 降級方法中:state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] count的值 = 18 執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 異常B，需要被記錄,方法被降級了~~ 降級方法中:state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=3, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行結束后：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=3, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 熔斷器已經打開：2019-07-29T17:36:14.189+08:00[Asia/Shanghai]: CircuitBreaker 'backendA' recorded a call which was not permitted. 執行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執行結束后：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 熔斷器已經打開，拒絕訪問被保護方法~

結果只有4個請求進去了，可以看出雖然熔斷器狀態還是半開，但是已經熔斷了，說明在半開狀態下，超過環大小的請求會被直接拒絕。

綜上，circuitbreaker的機制已經被證實，且十分清晰，以下為幾個需要注意的點：

失敗率的計算必須等環裝滿才會計算
白名單優先級高于黑名單且白名單上的異常會被忽略，不會占用緩沖環位置，即不會計入失敗率計算
熔斷器打開時同樣會計算失敗率，當狀態轉換為半開時重置為-1
只要出現異常都可以調用降級方法，不論是在白名單還是黑名單
熔斷器的緩沖環有兩個，一個關閉時的緩沖環，一個打開時的緩沖環
熔斷器關閉時，直至熔斷器狀態轉換前所有請求都會通過，不會受到限制
熔斷器半開時，限制請求數為緩沖環的大小，其他請求會等待
熔斷器從打開到半開的轉換默認還需要請求進行觸發，也可通過automaticTransitionFromOpenToHalfOpenEnabled=true設置為自動觸發

TimeLimiter

簡介

與Hystrix不同，Resilience4j將超時控制器從熔斷器中獨立出來，成為了一個單獨的組件，主要的作用就是對方法調用進行超時控制。實現的原理和Hystrix相似，都是通過調用Future的get方法來進行超時控制。

可配置參數

配置參數默認值描述

timeoutDuration	1(s)	超時時間限定
cancelRunningFuture	true	當超時時是否關閉取消線程

測試前準備

pom.xml

<dependency><groupId>io.github.resilience4j</groupId><artifactId>resilience4j-timelimiter</artifactId><version>0.16.0</version> </dependency>

TimeLimiter沒有整合進resilience4j-spring-boot中，需要單獨添加依賴

application.yml配置

timelimiter:timeoutDuration: 3000 # 超時時長cancelRunningFuture: true # 發生異常是否關閉線程

TimeLimiter沒有配置自動注入，需要自己進行注入，寫下面兩個文件進行配置自動注入：

TimeLimiterProperties

用于將application.yml中的配置轉換為TimeLimiterProperties對象：

@Data @Component @ConfigurationProperties(prefix = "resilience4j.timelimiter") public class TimeLimiterProperties {private Duration timeoutDuration;private boolean cancelRunningFuture; }

TimeLimiterConfiguration

將TimeLimiterProperties對象寫入到TimeLimiter的配置中：

@Configuration public class TimeLimiterConfiguration {@Autowiredprivate TimeLimiterProperties timeLimiterProperties;@Beanpublic TimeLimiter timeLimiter(){return TimeLimiter.of(timeLimiterConfig());}private TimeLimiterConfig timeLimiterConfig(){return TimeLimiterConfig.custom().timeoutDuration(timeLimiterProperties.getTimeoutDuration()).cancelRunningFuture(timeLimiterProperties.isCancelRunningFuture()).build();} }

調用方法

還是以之前查詢用戶列表的后端服務為例。TimeLimiter目前僅支持程序式調用，還不能使用AOP的方式調用。

因為TimeLimiter通常與CircuitBreaker聯合使用，很少單獨使用，所以直接介紹聯合使用的步驟。

TimeLimiter沒有注冊器，所以通過@Autowired注解自動注入依賴直接使用，因為TimeLimter是基于Future的get方法的，所以需要創建線程池，然后通過線程池的submit方法獲取Future對象：

public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;@Autowiredprivate TimeLimiter timeLimiter;public List<User> circuitBreakerTimeLimiter(){// 通過注冊器獲取熔斷器的實例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執行開始前：", circuitBreaker);// 創建單線程的線程池ExecutorService pool = Executors.newSingleThreadExecutor();//將被保護方法包裝為能夠返回Future的supplier函數Supplier<Future<List<User>>> futureSupplier = () -> pool.submit(remoteServiceConnector::process);// 先用限時器包裝，再用熔斷器包裝Callable<List<User>> restrictedCall = TimeLimiter.decorateFutureSupplier(timeLimiter, futureSupplier);Callable<List<User>> chainedCallable = CircuitBreaker.decorateCallable(circuitBreaker, restrictedCall);// 使用Try.of().recover()調用并進行降級處理Try<List<User>> result = Try.of(chainedCallable::call).recover(CallNotPermittedException.class, throwable ->{log.info("熔斷器已經打開，拒絕訪問被保護方法~");CircuitBreakerUtil.getCircuitBreakerStatus("熔斷器打開中", circuitBreaker);List<User> users = new ArrayList();return users;}).recover(throwable -> {log.info(throwable.getLocalizedMessage() + ",方法被降級了~~");CircuitBreakerUtil.getCircuitBreakerStatus("降級方法中:",circuitBreaker);List<User> users = new ArrayList();return users;});CircuitBreakerUtil.getCircuitBreakerStatus("執行結束后：", circuitBreaker);return result.get();} }

使用測試

異常A和B在application.yml文件中沒有修改：

使用另一個遠程服務接口的實現，將num%4==3的情況讓線程休眠5s，大于我們TimeLimiter的限制時間：

public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() {int num = count.getAndIncrement();log.info("count的值 = " + num);if (num % 4 == 1){throw new BusinessAException("異常A，不需要被記錄");}if (num % 4 == 2){throw new BusinessBException("異常B，需要被記錄");}if (num % 4 == 3){Thread.sleep(5000);}log.info("服務正常運行，獲取用戶列表");// 模擬數據庫的正常查詢return repository.findAll();} }

把調用方法進行單元測試，循環10遍：

public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerTimeLimiterTest() {for (int i=0; i<10; i++){circuitService.circuitBreakerTimeLimiter();}} }

看下運行結果：

發現熔斷器任何異常和超時都沒有失敗。。完全不會觸發熔斷，這是為什么呢？我們把異常toString()看一下：

java.util.concurrent.ExecutionException: com.example.resilience4j.exceptions.BusinessBException: 異常B，需要被記錄,方法被降級了~~ java.util.concurrent.TimeoutException,方法被降級了~~

這下原因就很明顯了，線程池會將線程中的任何異常包裝為ExecutionException，而熔斷器沒有把異常解包，由于我們設置了黑名單，而熔斷器又沒有找到黑名單上的異常，所以失效了。這是一個已知的bug，會在下個版本(0.16.0之后)中修正，目前來說如果需要同時使用TimeLimiter和CircuitBreaker的話，黑白名單的設置是不起作用的，需要自定義自己的謂詞邏輯，并在test()方法中將異常解包進行判斷，比如像下面這樣：

public class RecordFailurePredicate implements Predicate<Throwable> {@Overridepublic boolean test(Throwable throwable) {if (throwable.getCause() instanceof BusinessAException) return false;else return true;} }

然后在application.yml文件中指定這個類作為判斷類：

circuitbreaker:configs:default:recordFailurePredicate: com.example.resilience4j.predicate.RecordFailurePredicate

就能自定義自己的黑白名單了，我們再運行一次試試：

java.util.concurrent.TimeoutException,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執行結束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ]

可以看出，TimeLimiter已經生效了，同時CircuitBreaker也正常工作。

Note:

最新版0.17.0，該bug已經修復，黑白名單可以正常使用。

Retry

簡介

同熔斷器一樣，重試組件也提供了注冊器，可以通過注冊器獲取實例來進行重試，同樣可以跟熔斷器配合使用。

可配置參數

配置參數默認值描述

maxAttempts	3	最大重試次數
waitDuration	500[ms]	固定重試間隔
intervalFunction	numberOfAttempts -> waitDuration	用來改變重試時間間隔，可以選擇指數退避或者隨機時間間隔
retryOnResultPredicate	result -> false	自定義結果重試規則，需要重試的返回true
retryOnExceptionPredicate	throwable -> true	自定義異常重試規則，需要重試的返回true
retryExceptions	empty	需要重試的異常列表
ignoreExceptions	empty	需要忽略的異常列表

測試前準備

pom.xml

不需要引入新的依賴，已經集成在resilience4j-spring-boot中了

application.yml配置

resilience4j:retry:configs:default:maxRetryAttempts: 3waitDuration: 10senableExponentialBackoff: true # 是否允許使用指數退避算法進行重試間隔時間的計算expontialBackoffMultiplier: 2 # 指數退避算法的乘數enableRandomizedWait: false # 是否允許使用隨機的重試間隔randomizedWaitFactor: 0.5 # 隨機因子resultPredicate: com.example.resilience4j.predicate.RetryOnResultPredicate retryExceptionPredicate: com.example.resilience4j.predicate.RetryOnExceptionPredicateretryExceptions:- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAException- io.github.resilience4j.circuitbreaker.CallNotPermittedExceptionignoreExceptions:- io.github.resilience4j.circuitbreaker.CallNotPermittedExceptioninstances:backendA:baseConfig: defaultwaitDuration: 5sbackendB:baseConfig: defaultmaxRetryAttempts: 2

application.yml可以配置的參數多出了幾個enableExponentialBackoff、expontialBackoffMultiplier、enableRandomizedWait、randomizedWaitFactor，分別代表是否允許指數退避間隔時間，指數退避的乘數、是否允許隨機間隔時間、隨機因子，注意指數退避和隨機間隔不能同時啟用。

用于監控重試組件狀態及事件的工具類

同樣為了監控重試組件，寫一個工具類：

@Log4j2 public class RetryUtil {/*** @Description: 獲取重試的狀態*/public static void getRetryStatus(String time, Retry retry){Retry.Metrics metrics = retry.getMetrics();long failedRetryNum = metrics.getNumberOfFailedCallsWithRetryAttempt();long failedNotRetryNum = metrics.getNumberOfFailedCallsWithoutRetryAttempt();long successfulRetryNum = metrics.getNumberOfSuccessfulCallsWithRetryAttempt();long successfulNotyRetryNum = metrics.getNumberOfSuccessfulCallsWithoutRetryAttempt();log.info(time + "state=" + " metrics[ failedRetryNum=" + failedRetryNum +", failedNotRetryNum=" + failedNotRetryNum +", successfulRetryNum=" + successfulRetryNum +", successfulNotyRetryNum=" + successfulNotyRetryNum +" ]");}/*** @Description: 監聽重試事件*/public static void addRetryListener(Retry retry){retry.getEventPublisher().onSuccess(event -> log.info("服務調用成功：" + event.toString())).onError(event -> log.info("服務調用失敗：" + event.toString())).onIgnoredError(event -> log.info("服務調用失敗，但異常被忽略：" + event.toString())).onRetry(event -> log.info("重試：第" + event.getNumberOfRetryAttempts() + "次"));} }

調用方法

還是以之前查詢用戶列表的服務為例。Retry支持AOP和程序式兩種方式的調用.

程序式的調用方法

和CircuitBreaker的調用方式差不多，和熔斷器配合使用有兩種調用方式，一種是先用重試組件裝飾，再用熔斷器裝飾，這時熔斷器的失敗需要等重試結束才計算，另一種是先用熔斷器裝飾，再用重試組件裝飾，這時每次調用服務都會記錄進熔斷器的緩沖環中，需要注意的是，第二種方式需要把CallNotPermittedException放進重試組件的白名單中，因為熔斷器打開時重試是沒有意義的：

public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;@Autowiredprivate RetryRegistry retryRegistry;public List<User> circuitBreakerRetryNotAOP(){// 通過注冊器獲取熔斷器的實例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");// 通過注冊器獲取重試組件實例Retry retry = retryRegistry.retry("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執行開始前：", circuitBreaker);// 先用重試組件包裝，再用熔斷器包裝CheckedFunction0<List<User>> checkedSupplier = Retry.decorateCheckedSupplier(retry, remoteServiceConnector::process);CheckedFunction0<List<User>> chainedSupplier = CircuitBreaker .decorateCheckedSupplier(circuitBreaker, checkedSupplier);// 使用Try.of().recover()調用并進行降級處理Try<List<User>> result = Try.of(chainedSupplier).recover(CallNotPermittedException.class, throwable -> {log.info("已經被熔斷，停止重試");return new ArrayList<>();}).recover(throwable -> {log.info("重試失敗: " + throwable.getLocalizedMessage());return new ArrayList<>();});RetryUtil.getRetryStatus("執行結束: ", retry);CircuitBreakerUtil.getCircuitBreakerStatus("執行結束：", circuitBreaker);return result.get();} }

AOP式的調用方法

首先在連接器方法上使用@Retry(name="",fallbackMethod="")注解，其中name是要使用的重試器實例的名稱，fallbackMethod是要使用的降級方法：

public RemoteServiceConnector{@CircuitBreaker(name = "backendA", fallbackMethod = "fallBack")@Retry(name = "backendA", fallbackMethod = "fallBack")public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;} }

要求和熔斷器一致，但是需要注意同時注解重試組件和熔斷器的話，是按照第二種方案來的，即每一次請求都會被熔斷器記錄。

之后直接調用方法：

public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;@Autowiredprivate RetryRegistry retryRegistry;public List<User> circuitBreakerRetryAOP() throws TimeoutException, InterruptedException {List<User> result = remoteServiceConnector.process();RetryUtil.getRetryStatus("執行結束：", retryRegistry.retry("backendA"));CircuitBreakerUtil.getCircuitBreakerStatus("執行結束：", circuitBreakerRegistry.circuitBreaker("backendA"));return result;} }

使用測試

異常A和B在application.yml文件中設定為都需要重試，因為使用第一種方案，所以不需要將CallNotPermittedException設定在重試組件的白名單中，同時為了測試重試過程中的異常是否會被熔斷器記錄，將異常A從熔斷器白名單中去除：

recordExceptions: # 記錄的異常- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAException ignoreExceptions: # 忽略的異常 # - com.example.resilience4j.exceptions.BusinessAException # ... resultPredicate: com.example.resilience4j.predicate.RetryOnResultPredicate retryExceptions:- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAException- io.github.resilience4j.circuitbreaker.CallNotPermittedException ignoreExceptions: # - io.github.resilience4j.circuitbreaker.CallNotPermittedException

使用另一個遠程服務接口的實現，將num%4==2的情況返回null，測試根據返回結果進行重試的功能：

public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() {int num = count.getAndIncrement();log.info("count的值 = " + num);if (num % 4 == 1){throw new BusinessAException("異常A，需要重試");}if (num % 4 == 2){return null;}if (num % 4 == 3){throw new BusinessBException("異常B，需要重試");}log.info("服務正常運行，獲取用戶列表");// 模擬數據庫的正常查詢return repository.findAll();} }

同時添加一個類自定義哪些返回值需要重試，設定為返回值為空就進行重試，這樣num % 4 == 2時就可以測試不拋異常，根據返回結果進行重試了：

public class RetryOnResultPredicate implements Predicate {@Overridepublic boolean test(Object o) {return o == null ? true : false;} }

使用CircuitBreakerServiceImpl中的AOP或者程序式調用方法進行單元測試，循環調用10次：

public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerRetryTest() {for (int i=0; i<10; i++){// circuitService.circuitBreakerRetryAOP();circuitService.circuitBreakerRetryNotAOP();}} }

看一下運行結果：

count的值 = 0 服務正常運行，獲取用戶列表執行結束: state= metrics[ failedRetryNum=0, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=1 ] 執行結束：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 1 重試：第1次 count的值 = 2 重試：第2次 count的值 = 3 服務調用失敗：2019-07-09T19:06:59.705+08:00[Asia/Shanghai]: Retry 'backendA' recorded a failed retry attempt. Number of retry attempts: '3', Last exception was: 'com.example.resilience4j.exceptions.BusinessBException: 異常B，需要重試'. 重試失敗: 異常B，需要重試執行結束: state= metrics[ failedRetryNum=1, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=1 ] 執行結束：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ]

這部分結果可以看出來，重試最大次數設置為3結果其實只重試了2次，服務共執行了3次，重試3次后熔斷器只記錄了1次。而且返回值為null時也確實進行重試了。

服務正常運行，獲取用戶列表執行結束: state= metrics[ failedRetryNum=2, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=3 ] 執行結束：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=0 ] 已經被熔斷，停止重試執行結束: state= metrics[ failedRetryNum=2, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=3 ] 執行結束：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=1 ]

當熔斷之后不會再進行重試。

接下來我修改一下調用服務的實現：

public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() {int num = count.getAndIncrement();log.info("count的值 = " + num);if (num % 4 == 1){throw new BusinessAException("異常A，需要重試");}if (num % 4 == 3){return null;}if (num % 4 == 2){throw new BusinessBException("異常B，需要重試");}log.info("服務正常運行，獲取用戶列表");// 模擬數據庫的正常查詢return repository.findAll();} }

將num%4==2變成異常B，num%4==3變成返回null，看一下最后一次重試返回值為null屬于重試成功還是重試失敗。

運行結果如下：

如上可知如果最后一次重試不拋出異常就算作重試成功，不管結果是否需要繼續重試。

Bulkhead

簡介

Resilence4j的Bulkhead提供兩種實現，一種是基于信號量的，另一種是基于有等待隊列的固定大小的線程池的，由于基于信號量的Bulkhead能很好地在多線程和I/O模型下工作，所以選擇介紹基于信號量的Bulkhead的使用。

可配置參數

配置參數默認值描述

maxConcurrentCalls	25	可允許的最大并發線程數
maxWaitDuration	0	嘗試進入飽和艙壁時應阻止線程的最大時間

測試前準備

pom.xml

不需要引入新的依賴，已經集成在resilience4j-spring-boot中了

application.yml配置

resilience4j:bulkhead:configs:default:maxConcurrentCalls: 10maxWaitDuration: 1000instances:backendA:baseConfig: defaultmaxConcurrentCalls: 3backendB:baseConfig: defaultmaxWaitDuration: 100

和CircuitBreaker差不多，都是可以通過繼承覆蓋配置設定實例的。

用于監控Bulkhead狀態及事件的工具類

同樣為了監控Bulkhead組件，寫一個工具類：

@Log4j2 public class BulkhdadUtil {/*** @Description: 獲取bulkhead的狀態*/public static void getBulkheadStatus(String time, Bulkhead bulkhead){Bulkhead.Metrics metrics = bulkhead.getMetrics();// Returns the number of parallel executions this bulkhead can support at this point in time.int availableConcurrentCalls = metrics.getAvailableConcurrentCalls();// Returns the configured max amount of concurrent callsint maxAllowedConcurrentCalls = metrics.getMaxAllowedConcurrentCalls();log.info(time + ", metrics[ availableConcurrentCalls=" + availableConcurrentCalls +", maxAllowedConcurrentCalls=" + maxAllowedConcurrentCalls + " ]");}/*** @Description: 監聽bulkhead事件*/public static void addBulkheadListener(Bulkhead bulkhead){bulkhead.getEventPublisher().onCallFinished(event -> log.info(event.toString())).onCallPermitted(event -> log.info(event.toString())).onCallRejected(event -> log.info(event.toString()));} }

調用方法

還是以之前查詢用戶列表的服務為例。Bulkhead支持AOP和程序式兩種方式的調用。

程序式的調用方法

調用方法都類似，裝飾方法之后用Try.of().recover()來執行：

public class BulkheadServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate BulkheadRegistry bulkheadRegistry;public List<User> bulkheadNotAOP(){// 通過注冊器獲得Bulkhead實例Bulkhead bulkhead = bulkheadRegistry.bulkhead("backendA");BulkhdadUtil.getBulkheadStatus("開始執行前: ", bulkhead);// 通過Try.of().recover()調用裝飾后的服務Try<List<User>> result = Try.of(Bulkhead.decorateCheckedSupplier(bulkhead, remoteServiceConnector::process)).recover(BulkheadFullException.class, throwable -> {log.info("服務失敗: " + throwable.getLocalizedMessage());return new ArrayList();});BulkhdadUtil.getBulkheadStatus("執行結束: ", bulkhead);return result.get();} }

AOP式的調用方法

首先在連接器方法上使用@Bulkhead(name="", fallbackMethod="", type="")注解，其中name是要使用的Bulkhead實例的名稱，fallbackMethod是要使用的降級方法，type是選擇信號量或線程池的Bulkhead：

public RemoteServiceConnector{@Bulkhead(name = "backendA", fallbackMethod = "fallback", type = Bulkhead.Type.SEMAPHORE)public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;}private List<User> fallback(BulkheadFullException e){log.info("服務失敗: " + e.getLocalizedMessage());return new ArrayList();} }

如果Retry、CircuitBreaker、Bulkhead同時注解在方法上，默認的順序是Retry>CircuitBreaker>Bulkhead，即先控制并發再熔斷最后重試，之后直接調用方法：

public class BulkheadServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate BulkheadRegistry bulkheadRegistry;public List<User> bulkheadAOP() throws TimeoutException, InterruptedException {List<User> result = remoteServiceConnector.process();BulkheadUtil.getBulkheadStatus("執行結束：", bulkheadRegistry.retry("backendA"));return result;} }

使用測試

在application.yml文件中將backenA線程數限制為1，便于觀察，最大等待時間為1s，超過1s的會走降級方法：

instances:backendA:baseConfig: defaultmaxConcurrentCalls: 1

使用另一個遠程服務接口的實現，不拋出異常，當做正常服務進行：

用線程池調5個線程去請求服務：

public class BulkheadServiceImplTest{@Autowiredprivate BulkheadServiceImpl bulkheadService;@Autowiredprivate BulkheadRegistry bulkheadRegistry;@Testpublic void bulkheadTest() {BulkhdadUtil.addBulkheadListener(bulkheadRegistry.bulkhead("backendA"));ExecutorService pool = Executors.newCachedThreadPool();for (int i=0; i<5; i++){pool.submit(() -> {// bulkheadService.bulkheadAOP();bulkheadService.bulkheadNotAOP();});}pool.shutdown();while (!pool.isTerminated());}} }

看一下運行結果：

開始執行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' permitted a call. count的值 = 0 服務正常運行，獲取用戶列表開始執行前: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' rejected a call. Bulkhead 'backendA' rejected a call. Bulkhead 'backendA' rejected a call. Bulkhead 'backendA' rejected a call. 服務失敗: Bulkhead 'backendA' is full and does not permit further calls 執行結束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務失敗: Bulkhead 'backendA' is full and does not permit further calls 執行結束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務失敗: Bulkhead 'backendA' is full and does not permit further calls 執行結束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務失敗: Bulkhead 'backendA' is full and does not permit further calls 執行結束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' has finished a call. 執行結束: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ]

由上可以看出，5個請求只有一個進入，其余觸發rejected事件，然后自動進入降級方法。接下來我們把等待時間稍微加長一些：

instances:backendA:baseConfig: defaultmaxConcurrentCalls: 1maxWaitDuration: 5000

再運行一次：

前面的線程沒有馬上被拒絕，而是等待了一段時間再執行。

RateLimiter

簡介

高頻控制是可以限制服務調用頻率，Resilience4j的RateLimiter可以對頻率進行納秒級別的控制，在每一個周期刷新可以調用的次數，還可以設定線程等待權限的時間。

可配置參數

配置參數默認值描述

timeoutDuration	5[s]	線程等待權限的默認等待時間
limitRefreshPeriod	500[ns]	權限刷新的時間，每個周期結束后，RateLimiter將會把權限計數設置為limitForPeriod的值
limiteForPeriod	50	一個限制刷新期間的可用權限數

測試前準備

pom.xml

不需要引入新的依賴，已經集成在resilience4j-spring-boot中了

application.yml配置

resilience4j:ratelimiter:configs:default:limitForPeriod: 5limitRefreshPeriod: 1stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1backendB:baseConfig: defaulttimeoutDuration: 0s

用于監控RateLimiter狀態及事件的工具類

同樣為了監控RateLimiter組件，寫一個工具類：

@Log4j2 public class RateLimiterUtil {/*** @Description: 獲取rateLimiter的狀態*/public static void getRateLimiterStatus(String time, RateLimiter rateLimiter){RateLimiter.Metrics metrics = rateLimiter.getMetrics();// Returns the number of availablePermissions in this duration.int availablePermissions = metrics.getAvailablePermissions();// Returns the number of WaitingThreadsint numberOfWaitingThreads = metrics.getNumberOfWaitingThreads();log.info(time + ", metrics[ availablePermissions=" + availablePermissions +", numberOfWaitingThreads=" + numberOfWaitingThreads + " ]");}/*** @Description: 監聽rateLimiter事件*/public static void addRateLimiterListener(RateLimiter rateLimiter){rateLimiter.getEventPublisher().onSuccess(event -> log.info(event.toString())).onFailure(event -> log.info(event.toString()));} }

調用方法

還是以之前查詢用戶列表的服務為例。RateLimiter支持AOP和程序式兩種方式的調用。

程序式的調用方法

調用方法都類似，裝飾方法之后用Try.of().recover()來執行：

public class RateLimiterServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate RateLimiterRegistry rateLimiterRegistry;public List<User> ratelimiterNotAOP(){// 通過注冊器獲得RateLimiter實例RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("backendA");RateLimiterUtil.getRateLimiterStatus("開始執行前: ", rateLimiter);// 通過Try.of().recover()調用裝飾后的服務Try<List<User>> result = Try.of(Bulkhead.decorateCheckedSupplier(rateLimiter, remoteServiceConnector::process)).recover(BulkheadFullException.class, throwable -> {log.info("服務失敗: " + throwable.getLocalizedMessage());return new ArrayList();});RateLimiterUtil.getRateLimiterStatus("執行結束: ", rateLimiter);return result.get();} }

AOP式的調用方法

首先在連接器方法上使用@RateLimiter(name="", fallbackMethod="")注解，其中name是要使用的RateLimiter實例的名稱，fallbackMethod是要使用的降級方法：

public RemoteServiceConnector{@RateLimiter(name = "backendA", fallbackMethod = "fallback")public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;}private List<User> fallback(BulkheadFullException e){log.info("服務失敗: " + e.getLocalizedMessage());return new ArrayList();} }

如果Retry、CircuitBreaker、Bulkhead、RateLimiter同時注解在方法上，默認的順序是Retry>CircuitBreaker>RateLimiter>Bulkhead，即先控制并發再限流然后熔斷最后重試

接下來直接調用方法：

public class RateLimiterServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate RateLimiterRegistry rateLimiterRegistry;public List<User> rateLimiterAOP() throws TimeoutException, InterruptedException {List<User> result = remoteServiceConnector.process();BulkheadUtil.getBulkheadStatus("執行結束：", rateLimiterRegistry.retry("backendA"));return result;} }

使用測試

在application.yml文件中將backenA設定為20s只能處理1個請求，為便于觀察，刷新時間設定為20s，等待時間設定為5s：

configs:default:limitForPeriod: 5limitRefreshPeriod: 20stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1

使用另一個遠程服務接口的實現，不拋出異常，當做正常服務進行，為了讓結果明顯一些，讓方法sleep 5秒：

public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() throws InterruptedException {int num = count.getAndIncrement();log.info("count的值 = " + num);Thread.sleep(5000);log.info("服務正常運行，獲取用戶列表");// 模擬數據庫正常查詢return repository.findAll();} }

用線程池調5個線程去請求服務：

public class RateLimiterServiceImplTest{@Autowiredprivate RateLimiterServiceImpl rateLimiterService;@Autowiredprivate RateLimiterRegistry rateLimiterRegistry;@Testpublic void rateLimiterTest() {RateLimiterUtil.addRateLimiterListener(rateLimiterRegistry.rateLimiter("backendA"));ExecutorService pool = Executors.newCachedThreadPool();for (int i=0; i<5; i++){pool.submit(() -> {// rateLimiterService.rateLimiterAOP();rateLimiterService.rateLimiterNotAOP();});}pool.shutdown();while (!pool.isTerminated());}} }

看一下測試結果：

開始執行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=0, numberOfWaitingThreads=0 ] RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:15.735+08:00[Asia/Shanghai]} count的值 = 0 RateLimiterEvent{type=FAILED_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:20.737+08:00[Asia/Shanghai]} RateLimiterEvent{type=FAILED_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:20.739+08:00[Asia/Shanghai]} RateLimiterEvent{type=FAILED_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:20.740+08:00[Asia/Shanghai]} 服務失敗: RateLimiter 'backendA' does not permit further calls 服務失敗: RateLimiter 'backendA' does not permit further calls 執行結束: , metrics[ availablePermissions=0, numberOfWaitingThreads=1 ] 執行結束: , metrics[ availablePermissions=0, numberOfWaitingThreads=1 ] RateLimiterEvent{type=FAILED_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:20.745+08:00[Asia/Shanghai]} 服務正常運行，獲取用戶列表服務失敗: RateLimiter 'backendA' does not permit further calls 執行結束: , metrics[ availablePermissions=0, numberOfWaitingThreads=0 ] 服務失敗: RateLimiter 'backendA' does not permit further calls 執行結束: , metrics[ availablePermissions=0, numberOfWaitingThreads=0 ] 執行結束: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ]

只有一個服務調用成功，其他都執行失敗了?，F在我們把刷新時間調成1s：

configs:default:limitForPeriod: 5limitRefreshPeriod: 1stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1

重新執行，結果如下：

開始執行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T18:25:18.894+08:00[Asia/Shanghai]}count的值 = 0 RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T18:25:18.894+08:00[Asia/Shanghai]} count的值 = 1 RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T18:25:19.706+08:00[Asia/Shanghai]} count的值 = 2 RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T18:25:19.706+08:00[Asia/Shanghai]} count的值 = 3 RateLimiterEvent{type=SUCCESSFUL_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T18:25:20.703+08:00[Asia/Shanghai]} count的值 = 4 服務正常運行，獲取用戶列表服務正常運行，獲取用戶列表服務正常運行，獲取用戶列表服務正常運行，獲取用戶列表執行結束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 執行結束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 執行結束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ]執行結束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 服務正常運行，獲取用戶列表執行結束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ]

可以看出，幾個服務都被放入并正常執行了，即使上個服務還沒完成，依然可以放入，只與時間有關，而與線程無關。

總結

以上是生活随笔為你收集整理的Resilience4j-轻量级熔断框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： JSR 303 - Bean Valid
下一篇： Zookeeper集群脑裂问题