Resilience4j-轻量级熔断框架
生活随笔
收集整理的這篇文章主要介紹了
Resilience4j-轻量级熔断框架
小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
Resilience4j
簡介
Resilience4j是一款輕量級，易于使用的容錯(cuò)庫，其靈感來自于Netflix Hystrix，但是專為Java 8和函數(shù)式編程而設(shè)計(jì)。輕量級，因?yàn)閹熘皇褂昧?strong>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通過具有三種正常狀態(tài)的有限狀態(tài)機(jī)實(shí)現(xiàn)：CLOSED，OPEN和HALF_OPEN以及兩個(gè)特殊狀態(tài)DISABLED和FORCED_OPEN。當(dāng)熔斷器關(guān)閉時(shí)，所有的請求都會(huì)通過熔斷器。如果失敗率超過設(shè)定的閾值，熔斷器就會(huì)從關(guān)閉狀態(tài)轉(zhuǎn)換到打開狀態(tài)，這時(shí)所有的請求都會(huì)被拒絕。當(dāng)經(jīng)過一段時(shí)間后，熔斷器會(huì)從打開狀態(tài)轉(zhuǎn)換到半開狀態(tài)，這時(shí)僅有一定數(shù)量的請求會(huì)被放入，并重新計(jì)算失敗率，如果失敗率超過閾值，則變?yōu)榇蜷_狀態(tài)，如果失敗率低于閾值，則變?yōu)殛P(guān)閉狀態(tài)。
Circuitbreaker狀態(tài)機(jī)
Resilience4j記錄請求狀態(tài)的數(shù)據(jù)結(jié)構(gòu)和Hystrix不同，Hystrix是使用滑動(dòng)窗口來進(jìn)行存儲的，而Resilience4j采用的是Ring Bit Buffer(環(huán)形緩沖區(qū))。Ring Bit Buffer在內(nèi)部使用BitSet這樣的數(shù)據(jù)結(jié)構(gòu)來進(jìn)行存儲，BitSet的結(jié)構(gòu)如下圖所示：
環(huán)形緩沖區(qū)
每一次請求的成功或失敗狀態(tài)只占用一個(gè)bit位，與boolean數(shù)組相比更節(jié)省內(nèi)存。BitSet使用long[]數(shù)組來存儲這些數(shù)據(jù)，意味著16個(gè)值(64bit)的數(shù)組可以存儲1024個(gè)調(diào)用狀態(tài)。
計(jì)算失敗率需要填滿環(huán)形緩沖區(qū)。例如，如果環(huán)形緩沖區(qū)的大小為10，則必須至少請求滿10次，才會(huì)進(jìn)行故障率的計(jì)算，如果僅僅請求了9次，即使9個(gè)請求都失敗，熔斷器也不會(huì)打開。但是CLOSE狀態(tài)下的緩沖區(qū)大小設(shè)置為10并不意味著只會(huì)進(jìn)入10個(gè) 請求，在熔斷器打開之前的所有請求都會(huì)被放入。
當(dāng)故障率高于設(shè)定的閾值時(shí)，熔斷器狀態(tài)會(huì)從由CLOSE變?yōu)?strong>OPEN。這時(shí)所有的請求都會(huì)拋出CallNotPermittedException異常。當(dāng)經(jīng)過一段時(shí)間后，熔斷器的狀態(tài)會(huì)從OPEN變?yōu)?strong>HALF_OPEN，HALF_OPEN狀態(tài)下同樣會(huì)有一個(gè)Ring Bit Buffer，用來計(jì)算HALF_OPEN狀態(tài)下的故障率，如果高于配置的閾值，會(huì)轉(zhuǎn)換為OPEN，低于閾值則裝換為CLOSE。與CLOSE狀態(tài)下的緩沖區(qū)不同的地方在于，HALF_OPEN狀態(tài)下的緩沖區(qū)大小會(huì)限制請求數(shù)，只有緩沖區(qū)大小的請求數(shù)會(huì)被放入。
除此以外，熔斷器還會(huì)有兩種特殊狀態(tài)：DISABLED（始終允許訪問）和FORCED_OPEN（始終拒絕訪問）。這兩個(gè)狀態(tài)不會(huì)生成熔斷器事件（除狀態(tài)裝換外），并且不會(huì)記錄事件的成功或者失敗。退出這兩個(gè)狀態(tài)的唯一方法是觸發(fā)狀態(tài)轉(zhuǎn)換或者重置熔斷器。
熔斷器關(guān)于線程安全的保證措施有以下幾個(gè)部分：
- 熔斷器的狀態(tài)使用AtomicReference保存的
- 更新熔斷器狀態(tài)是通過無狀態(tài)的函數(shù)或者原子操作進(jìn)行的
- 更新事件的狀態(tài)用synchronized關(guān)鍵字保護(hù)
意味著同一時(shí)間只有一個(gè)線程能夠修改熔斷器狀態(tài)或者記錄事件的狀態(tài)。
可配置參數(shù)
| failureRateThreshold | 50 | 熔斷器關(guān)閉狀態(tài)和半開狀態(tài)使用的同一個(gè)失敗率閾值 |
| ringBufferSizeInHalfOpenState | 10 | 熔斷器半開狀態(tài)的緩沖區(qū)大小，會(huì)限制線程的并發(fā)量，例如緩沖區(qū)為10則每次只會(huì)允許10個(gè)請求調(diào)用后端服務(wù) |
| ringBufferSizeInClosedState | 100 | 熔斷器關(guān)閉狀態(tài)的緩沖區(qū)大小，不會(huì)限制線程的并發(fā)量，在熔斷器發(fā)生狀態(tài)轉(zhuǎn)換前所有請求都會(huì)調(diào)用后端服務(wù) |
| waitDurationInOpenState | 60(s) | 熔斷器從打開狀態(tài)轉(zhuǎn)變?yōu)榘腴_狀態(tài)等待的時(shí)間 |
| automaticTransitionFromOpenToHalfOpenEnabled | false | 如果置為true，當(dāng)?shù)却龝r(shí)間結(jié)束會(huì)自動(dòng)由打開變?yōu)榘腴_，若置為false，則需要一個(gè)請求進(jìn)入來觸發(fā)熔斷器狀態(tài)轉(zhuǎn)換 |
| recordExceptions | empty | 需要記錄為失敗的異常列表 |
| ignoreExceptions | empty | 需要忽略的異常列表 |
| recordFailure | throwable -> true | 自定義的謂詞邏輯用于判斷異常是否需要記錄或者需要忽略，默認(rèn)所有異常都進(jìn)行記錄 |
測試前準(zhǔn)備
pom.xml
測試使用的IDE為idea，使用的springboot進(jìn)行學(xué)習(xí)測試，首先引入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幾個(gè)模塊，因?yàn)楹罄m(xù)還要學(xué)習(xí)其他模塊，就直接引入resilience4j-spring-boot依賴。
application.yml配置
?
resilience4j:circuitbreaker:configs:default:ringBufferSizeInClosedState: 5 # 熔斷器關(guān)閉時(shí)的緩沖區(qū)大小ringBufferSizeInHalfOpenState: 2 # 熔斷器半開時(shí)的緩沖區(qū)大小waitDurationInOpenState: 10000 # 熔斷器從打開到半開需要的時(shí)間failureRateThreshold: 60 # 熔斷器打開的失敗閾值eventConsumerBufferSize: 10 # 事件緩沖區(qū)大小registerHealthIndicator: true # 健康監(jiān)測automaticTransitionFromOpenToHalfOpenEnabled: false # 是否自動(dòng)從打開到半開，不需要觸發(fā)recordFailurePredicate: com.example.resilience4j.exceptions.RecordFailurePredicate # 謂詞設(shè)置異常是否為失敗recordExceptions: # 記錄的異常- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAExceptionignoreExceptions: # 忽略的異常- com.example.resilience4j.exceptions.BusinessAExceptioninstances:backendA:baseConfig: defaultwaitDurationInOpenState: 5000failureRateThreshold: 20backendB:baseConfig: default可以配置多個(gè)熔斷器實(shí)例，使用不同配置或者覆蓋配置。
需要保護(hù)的后端服務(wù)
以一個(gè)查找用戶列表的后端服務(wù)為例，利用熔斷器保護(hù)該服務(wù)。
?
interface RemoteService {List<User> process() throws TimeoutException, InterruptedException; }連接器調(diào)用該服務(wù)
這是調(diào)用遠(yuǎn)端服務(wù)的連接器，我們通過調(diào)用連接器中的方法來調(diào)用后端服務(wù)。
?
public RemoteServiceConnector{public List<User> process() throws TimeoutException, InterruptedException {List<User> users;users = remoteServic.process();return users;} }用于監(jiān)控熔斷器狀態(tài)及事件的工具類
要想學(xué)習(xí)各個(gè)配置項(xiàng)的作用，需要獲取特定時(shí)候的熔斷器狀態(tài)，寫一個(gè)工具類：
?
@Log4j2 public class CircuitBreakerUtil {/*** @Description: 獲取熔斷器的狀態(tài)*/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: 監(jiān)聽熔斷器事件*/public static void addCircuitBreakerListener(CircuitBreaker circuitBreaker){circuitBreaker.getEventPublisher().onSuccess(event -> log.info("服務(wù)調(diào)用成功：" + event.toString())).onError(event -> log.info("服務(wù)調(diào)用失敗：" + event.toString())).onIgnoredError(event -> log.info("服務(wù)調(diào)用失敗，但異常被忽略：" + event.toString())).onReset(event -> log.info("熔斷器重置：" + event.toString())).onStateTransition(event -> log.info("熔斷器狀態(tài)改變：" + event.toString())).onCallNotPermitted(event -> log.info(" 熔斷器已經(jīng)打開：" + event.toString()));}調(diào)用方法
CircuitBreaker目前支持兩種方式調(diào)用，一種是程序式調(diào)用，一種是AOP使用注解的方式調(diào)用。
程序式的調(diào)用方法
在CircuitService中先注入注冊器，然后用注冊器通過熔斷器名稱獲取熔斷器。如果不需要使用降級函數(shù)，可以直接調(diào)用熔斷器的executeSupplier方法或executeCheckedSupplier方法：
?
public class CircuitBreakerServiceImpl{@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerNotAOP() throws Throwable {CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行開始前：", circuitBreaker);circuitBreaker.executeCheckedSupplier(remotServiceConnector::process);} }如果需要使用降級函數(shù)，則要使用decorate包裝服務(wù)的方法，再使用Try.of().recover()進(jìn)行降級處理，同時(shí)也可以根據(jù)不同的異常使用不同的降級方法：
?
public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerNotAOP(){// 通過注冊器獲取熔斷器的實(shí)例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行開始前：", circuitBreaker);// 使用熔斷器包裝連接器的方法CheckedFunction0<List<User>> checkedSupplier = CircuitBreaker.decorateCheckedSupplier(circuitBreaker, remoteServiceConnector::process);// 使用Try.of().recover()調(diào)用并進(jìn)行降級處理Try<List<User>> result = Try.of(checkedSupplier).recover(CallNotPermittedException.class, throwable -> {log.info("熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~");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("執(zhí)行結(jié)束后：", circuitBreaker);return result.get();} }AOP式的調(diào)用方法
首先在連接器方法上使用@CircuitBreaker(name="",fallbackMethod="")注解，其中name是要使用的熔斷器的名稱，fallbackMethod是要使用的降級方法，降級方法必須和原方法放在同一個(gè)類中，且降級方法的返回值需要和原方法相同，輸入?yún)?shù)需要添加額外的exception參數(shù)，類似這樣：
?
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("熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~");CircuitBreakerUtil.getCircuitBreakerStatus("熔斷器打開中:", circuitBreakerRegistry.circuitBreaker("backendA"));List<User> users = new ArrayList();return users;}}可使用多個(gè)降級方法，保持方法名相同，同時(shí)滿足的條件的降級方法會(huì)觸發(fā)最接近的一個(gè)（這里的接近是指類型的接近，先會(huì)觸發(fā)離它最近的子類異常），例如如果process()方法拋出CallNotPermittedException，將會(huì)觸發(fā)fallBack(CallNotPermittedException e)方法而不會(huì)觸發(fā)fallBack(Throwable throwable)方法。
之后直接調(diào)用方法就可以了：
?
public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;public List<User> circuitBreakerAOP() throws TimeoutException, InterruptedException {CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行開始前：",circuitBreakerRegistry.circuitBreaker("backendA"));List<User> result = remoteServiceConnector.process();CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行結(jié)束后：", circuitBreakerRegistry.circuitBreaker("backendA"));return result;} }使用測試
接下來進(jìn)入測試，首先我們定義了兩個(gè)異常，異常A同時(shí)在黑白名單中，異常B只在黑名單中：
?
recordExceptions: # 記錄的異常- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAException ignoreExceptions: # 忽略的異常- com.example.resilience4j.exceptions.BusinessAException然后對被保護(hù)的后端接口進(jìn)行如下的實(shí)現(xiàn)：
?
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("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫的正常查詢r(jià)eturn repository.findAll();} }使用CircuitBreakerServiceImpl中的AOP或者程序式調(diào)用方法進(jìn)行單元測試，循環(huán)調(diào)用10次：
?
public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerTest() {for (int i=0; i<10; i++){// circuitService.circuitBreakerAOP();circuitService.circuitBreakerNotAOP();}} }看下運(yùn)行結(jié)果：
?
執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 4 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=4, failedCalls=2, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~ 熔斷器打開中:state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ]注意到異常A發(fā)生的前后bufferedCalls、failedCalls、successCalls三個(gè)參數(shù)的值都沒有沒有發(fā)生變化，說明白名單的優(yōu)先級高于黑名單，源碼中也有提到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> */同時(shí)也可以看出白名單所謂的忽略，是指不計(jì)入緩沖區(qū)中（即不算成功也不算失敗），有降級方法會(huì)調(diào)用降級方法，沒有降級方法會(huì)拋出異常，和其他異常無異。
?
執(zhí)行開始前：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~ 熔斷器打開中:state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=1 ]當(dāng)環(huán)形緩沖區(qū)大小被填滿時(shí)會(huì)計(jì)算失敗率，這時(shí)請求會(huì)被拒絕獲取不到count的值，且notPermittedCalls會(huì)增加。
接下來我們實(shí)驗(yàn)一下多線程下熔斷器關(guān)閉和熔斷器半開兩種情況下緩沖環(huán)的區(qū)別，我們先開15個(gè)線程進(jìn)行調(diào)用測試熔斷器關(guān)閉時(shí)的緩沖環(huán)，熔斷之后等10s再開15個(gè)線程進(jìn)行調(diào)用測試熔斷器半開時(shí)的緩沖環(huán)：
?
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("熔斷器狀態(tài)已轉(zhuǎn)為半開");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個(gè)線程都通過了熔斷器，由于正常返回需要查數(shù)據(jù)庫，所以會(huì)慢很多，失敗率很快就達(dá)到了100%，而且觀察到如下的記錄：
?
異常B，需要被記錄,方法被降級了~~ 降級方法中:state=OPEN , metrics[ failureRate=100.0, bufferedCalls=5, failedCalls=5, successCalls=0, maxBufferCalls=5, notPermittedCalls=0 ]可以看出，雖然熔斷器已經(jīng)打開了，可是異常B還是進(jìn)入了降級方法，拋出的異常不是notPermittedCalls數(shù)量為0，說明在熔斷器轉(zhuǎn)換成打開之前所有請求都通過了熔斷器，緩沖環(huán)不會(huì)控制線程的并發(fā)。
?
執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=80.0, bufferedCalls=5, failedCalls=4, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=60.0, bufferedCalls=5, failedCalls=3, successCalls=2, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=OPEN , metrics[ failureRate=20.0, bufferedCalls=5, failedCalls=1, successCalls=4, maxBufferCalls=5, notPermittedCalls=0 ]同時(shí)以上幾條正常執(zhí)行的服務(wù)完成后，熔斷器的失敗率在下降，說明熔斷器打開狀態(tài)下還是會(huì)計(jì)算失敗率，由于環(huán)形緩沖區(qū)大小為5，初步推斷成功的狀態(tài)會(huì)依次覆蓋最開始的幾個(gè)狀態(tài)，所以得到了上述結(jié)果。
接下來分析后15個(gè)線程的結(jié)果
?
熔斷器狀態(tài)已轉(zhuǎn)為半開 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 16 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行開始前：state=OPEN , metrics[ failureRate=0.0, bufferedCalls=5, failedCalls=0, successCalls=5, maxBufferCalls=5, notPermittedCalls=0 ] 熔斷器狀態(tài)改變：2019-07-29T17:19:19.959+08:00[Asia/Shanghai]: CircuitBreaker 'backendA' changed state from OPEN to HALF_OPEN count的值 = 18 count的值 = 17 服務(wù)正常運(yùn)行，獲取用戶列表 count的值 = 19 count的值 = 15熔斷器狀態(tài)從打開到半開我設(shè)置的是5s，前15個(gè)線程調(diào)用之后我等待了10s，熔斷器應(yīng)該已經(jīng)變?yōu)榘腴_了，但是執(zhí)行開始前熔斷器的狀態(tài)卻是OPEN，這是因?yàn)槟J(rèn)的配置項(xiàng)automaticTransitionFromOpenToHalfOpenEnabled=false，時(shí)間到了也不會(huì)自動(dòng)轉(zhuǎn)換，需要有新的請求來觸發(fā)熔斷器的狀態(tài)轉(zhuǎn)換。同時(shí)我們發(fā)現(xiàn)，好像狀態(tài)改變后還是進(jìn)了超過4個(gè)請求，似乎半開狀態(tài)的環(huán)并不能限制線程數(shù)？這是由于這些進(jìn)程是在熔斷器打開時(shí)一起進(jìn)來的。為了更好的觀察環(huán)半開時(shí)候環(huán)大小是否限制線程數(shù)，我們修改一下配置：
?
resilience4j:circuitbreaker:configs:myDefault:automaticTransitionFromOpenToHalfOpenEnabled: true # 是否自動(dòng)從打開到半開我們再試一次：
?
熔斷器狀態(tài)已轉(zhuǎn)為半開 執(zhí)行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執(zhí)行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執(zhí)行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] count的值 = 15 count的值 = 16 服務(wù)正常運(yùn)行，獲取用戶列表異常B，需要被記錄,方法被降級了~~ 降級方法中:state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=1, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：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 ] 執(zhí)行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] count的值 = 18 執(zhí)行開始前：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 ] 執(zhí)行結(jié)束后：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=3, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 熔斷器已經(jīng)打開：2019-07-29T17:36:14.189+08:00[Asia/Shanghai]: CircuitBreaker 'backendA' recorded a call which was not permitted. 執(zhí)行開始前：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=HALF_OPEN , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=2, successCalls=0, maxBufferCalls=4, notPermittedCalls=0 ] 熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~結(jié)果只有4個(gè)請求進(jìn)去了，可以看出雖然熔斷器狀態(tài)還是半開，但是已經(jīng)熔斷了，說明在半開狀態(tài)下，超過環(huán)大小的請求會(huì)被直接拒絕。
綜上，circuitbreaker的機(jī)制已經(jīng)被證實(shí)，且十分清晰，以下為幾個(gè)需要注意的點(diǎn)：
- 失敗率的計(jì)算必須等環(huán)裝滿才會(huì)計(jì)算
- 白名單優(yōu)先級高于黑名單且白名單上的異常會(huì)被忽略，不會(huì)占用緩沖環(huán)位置，即不會(huì)計(jì)入失敗率計(jì)算
- 熔斷器打開時(shí)同樣會(huì)計(jì)算失敗率，當(dāng)狀態(tài)轉(zhuǎn)換為半開時(shí)重置為-1
- 只要出現(xiàn)異常都可以調(diào)用降級方法，不論是在白名單還是黑名單
- 熔斷器的緩沖環(huán)有兩個(gè)，一個(gè)關(guān)閉時(shí)的緩沖環(huán)，一個(gè)打開時(shí)的緩沖環(huán)
- 熔斷器關(guān)閉時(shí)，直至熔斷器狀態(tài)轉(zhuǎn)換前所有請求都會(huì)通過，不會(huì)受到限制
- 熔斷器半開時(shí)，限制請求數(shù)為緩沖環(huán)的大小，其他請求會(huì)等待
- 熔斷器從打開到半開的轉(zhuǎn)換默認(rèn)還需要請求進(jìn)行觸發(fā)，也可通過automaticTransitionFromOpenToHalfOpenEnabled=true設(shè)置為自動(dòng)觸發(fā)
TimeLimiter
簡介
與Hystrix不同，Resilience4j將超時(shí)控制器從熔斷器中獨(dú)立出來，成為了一個(gè)單獨(dú)的組件，主要的作用就是對方法調(diào)用進(jìn)行超時(shí)控制。實(shí)現(xiàn)的原理和Hystrix相似，都是通過調(diào)用Future的get方法來進(jìn)行超時(shí)控制。
可配置參數(shù)
| timeoutDuration | 1(s) | 超時(shí)時(shí)間限定 |
| cancelRunningFuture | true | 當(dāng)超時(shí)時(shí)是否關(guān)閉取消線程 |
測試前準(zhǔn)備
pom.xml
?
<dependency><groupId>io.github.resilience4j</groupId><artifactId>resilience4j-timelimiter</artifactId><version>0.16.0</version> </dependency>TimeLimiter沒有整合進(jìn)resilience4j-spring-boot中，需要單獨(dú)添加依賴
application.yml配置
?
timelimiter:timeoutDuration: 3000 # 超時(shí)時(shí)長cancelRunningFuture: true # 發(fā)生異常是否關(guān)閉線程TimeLimiter沒有配置自動(dòng)注入，需要自己進(jìn)行注入，寫下面兩個(gè)文件進(jìn)行配置自動(dòng)注入：
TimeLimiterProperties
用于將application.yml中的配置轉(zhuǎn)換為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();} }調(diào)用方法
還是以之前查詢用戶列表的后端服務(wù)為例。TimeLimiter目前僅支持程序式調(diào)用，還不能使用AOP的方式調(diào)用。
因?yàn)?strong>TimeLimiter通常與CircuitBreaker聯(lián)合使用，很少單獨(dú)使用，所以直接介紹聯(lián)合使用的步驟。
TimeLimiter沒有注冊器，所以通過@Autowired注解自動(dòng)注入依賴直接使用，因?yàn)?strong>TimeLimter是基于Future的get方法的，所以需要?jiǎng)?chuàng)建線程池，然后通過線程池的submit方法獲取Future對象：
?
public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;@Autowiredprivate TimeLimiter timeLimiter;public List<User> circuitBreakerTimeLimiter(){// 通過注冊器獲取熔斷器的實(shí)例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行開始前：", circuitBreaker);// 創(chuàng)建單線程的線程池ExecutorService pool = Executors.newSingleThreadExecutor();//將被保護(hù)方法包裝為能夠返回Future的supplier函數(shù)Supplier<Future<List<User>>> futureSupplier = () -> pool.submit(remoteServiceConnector::process);// 先用限時(shí)器包裝，再用熔斷器包裝Callable<List<User>> restrictedCall = TimeLimiter.decorateFutureSupplier(timeLimiter, futureSupplier);Callable<List<User>> chainedCallable = CircuitBreaker.decorateCallable(circuitBreaker, restrictedCall);// 使用Try.of().recover()調(diào)用并進(jìn)行降級處理Try<List<User>> result = Try.of(chainedCallable::call).recover(CallNotPermittedException.class, throwable ->{log.info("熔斷器已經(jīng)打開，拒絕訪問被保護(hù)方法~");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("執(zhí)行結(jié)束后：", circuitBreaker);return result.get();} }使用測試
異常A和B在application.yml文件中沒有修改：
?
recordExceptions: # 記錄的異常- com.example.resilience4j.exceptions.BusinessBException- com.example.resilience4j.exceptions.BusinessAException ignoreExceptions: # 忽略的異常- com.example.resilience4j.exceptions.BusinessAException使用另一個(gè)遠(yuǎn)程服務(wù)接口的實(shí)現(xiàn)，將num%4==3的情況讓線程休眠5s，大于我們TimeLimiter的限制時(shí)間：
?
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("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫的正常查詢r(jià)eturn repository.findAll();} }把調(diào)用方法進(jìn)行單元測試，循環(huán)10遍：
?
public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerTimeLimiterTest() {for (int i=0; i<10; i++){circuitService.circuitBreakerTimeLimiter();}} }看下運(yùn)行結(jié)果：
?
執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=0, failedCalls=0, successCalls=0, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 1 com.example.resilience4j.exceptions.BusinessAException: 異常A，不需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 2 com.example.resilience4j.exceptions.BusinessBException: 異常B，需要被記錄,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行開始前：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 3 null,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ]發(fā)現(xiàn)熔斷器任何異常和超時(shí)都沒有失敗。。完全不會(huì)觸發(fā)熔斷，這是為什么呢？我們把異常toString()看一下：
?
java.util.concurrent.ExecutionException: com.example.resilience4j.exceptions.BusinessBException: 異常B，需要被記錄,方法被降級了~~ java.util.concurrent.TimeoutException,方法被降級了~~這下原因就很明顯了，線程池會(huì)將線程中的任何異常包裝為ExecutionException，而熔斷器沒有把異常解包，由于我們設(shè)置了黑名單，而熔斷器又沒有找到黑名單上的異常，所以失效了。這是一個(gè)已知的bug，會(huì)在下個(gè)版本(0.16.0之后)中修正，目前來說如果需要同時(shí)使用TimeLimiter和CircuitBreaker的話，黑白名單的設(shè)置是不起作用的，需要自定義自己的謂詞邏輯，并在test()方法中將異常解包進(jìn)行判斷，比如像下面這樣：
?
public class RecordFailurePredicate implements Predicate<Throwable> {@Overridepublic boolean test(Throwable throwable) {if (throwable.getCause() instanceof BusinessAException) return false;else return true;} }然后在application.yml文件中指定這個(gè)類作為判斷類：
?
circuitbreaker:configs:default:recordFailurePredicate: com.example.resilience4j.predicate.RecordFailurePredicate就能自定義自己的黑白名單了，我們再運(yùn)行一次試試：
?
java.util.concurrent.TimeoutException,方法被降級了~~ 降級方法中:state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] 執(zhí)行結(jié)束后：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=3, failedCalls=2, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ]可以看出，TimeLimiter已經(jīng)生效了，同時(shí)CircuitBreaker也正常工作。
Note:
最新版0.17.0，該bug已經(jīng)修復(fù)，黑白名單可以正常使用。
Retry
簡介
同熔斷器一樣，重試組件也提供了注冊器，可以通過注冊器獲取實(shí)例來進(jìn)行重試，同樣可以跟熔斷器配合使用。
可配置參數(shù)
| maxAttempts | 3 | 最大重試次數(shù) |
| waitDuration | 500[ms] | 固定重試間隔 |
| intervalFunction | numberOfAttempts -> waitDuration | 用來改變重試時(shí)間間隔，可以選擇指數(shù)退避或者隨機(jī)時(shí)間間隔 |
| retryOnResultPredicate | result -> false | 自定義結(jié)果重試規(guī)則，需要重試的返回true |
| retryOnExceptionPredicate | throwable -> true | 自定義異常重試規(guī)則，需要重試的返回true |
| retryExceptions | empty | 需要重試的異常列表 |
| ignoreExceptions | empty | 需要忽略的異常列表 |
測試前準(zhǔn)備
pom.xml
不需要引入新的依賴，已經(jīng)集成在resilience4j-spring-boot中了
application.yml配置
?
resilience4j:retry:configs:default:maxRetryAttempts: 3waitDuration: 10senableExponentialBackoff: true # 是否允許使用指數(shù)退避算法進(jìn)行重試間隔時(shí)間的計(jì)算expontialBackoffMultiplier: 2 # 指數(shù)退避算法的乘數(shù)enableRandomizedWait: false # 是否允許使用隨機(jī)的重試間隔randomizedWaitFactor: 0.5 # 隨機(jī)因子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: 2application.yml可以配置的參數(shù)多出了幾個(gè)enableExponentialBackoff、expontialBackoffMultiplier、enableRandomizedWait、randomizedWaitFactor，分別代表是否允許指數(shù)退避間隔時(shí)間，指數(shù)退避的乘數(shù)、是否允許隨機(jī)間隔時(shí)間、隨機(jī)因子，注意指數(shù)退避和隨機(jī)間隔不能同時(shí)啟用。
用于監(jiān)控重試組件狀態(tài)及事件的工具類
同樣為了監(jiān)控重試組件，寫一個(gè)工具類：
?
@Log4j2 public class RetryUtil {/*** @Description: 獲取重試的狀態(tài)*/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: 監(jiān)聽重試事件*/public static void addRetryListener(Retry retry){retry.getEventPublisher().onSuccess(event -> log.info("服務(wù)調(diào)用成功：" + event.toString())).onError(event -> log.info("服務(wù)調(diào)用失敗：" + event.toString())).onIgnoredError(event -> log.info("服務(wù)調(diào)用失敗，但異常被忽略：" + event.toString())).onRetry(event -> log.info("重試：第" + event.getNumberOfRetryAttempts() + "次"));} }調(diào)用方法
還是以之前查詢用戶列表的服務(wù)為例。Retry支持AOP和程序式兩種方式的調(diào)用.
程序式的調(diào)用方法
和CircuitBreaker的調(diào)用方式差不多，和熔斷器配合使用有兩種調(diào)用方式，一種是先用重試組件裝飾，再用熔斷器裝飾，這時(shí)熔斷器的失敗需要等重試結(jié)束才計(jì)算，另一種是先用熔斷器裝飾，再用重試組件裝飾，這時(shí)每次調(diào)用服務(wù)都會(huì)記錄進(jìn)熔斷器的緩沖環(huán)中，需要注意的是，第二種方式需要把CallNotPermittedException放進(jìn)重試組件的白名單中，因?yàn)槿蹟嗥鞔蜷_時(shí)重試是沒有意義的：
?
public class CircuitBreakerServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate CircuitBreakerRegistry circuitBreakerRegistry;@Autowiredprivate RetryRegistry retryRegistry;public List<User> circuitBreakerRetryNotAOP(){// 通過注冊器獲取熔斷器的實(shí)例CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("backendA");// 通過注冊器獲取重試組件實(shí)例Retry retry = retryRegistry.retry("backendA");CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行開始前：", circuitBreaker);// 先用重試組件包裝，再用熔斷器包裝CheckedFunction0<List<User>> checkedSupplier = Retry.decorateCheckedSupplier(retry, remoteServiceConnector::process);CheckedFunction0<List<User>> chainedSupplier = CircuitBreaker .decorateCheckedSupplier(circuitBreaker, checkedSupplier);// 使用Try.of().recover()調(diào)用并進(jìn)行降級處理Try<List<User>> result = Try.of(chainedSupplier).recover(CallNotPermittedException.class, throwable -> {log.info("已經(jīng)被熔斷，停止重試");return new ArrayList<>();}).recover(throwable -> {log.info("重試失敗: " + throwable.getLocalizedMessage());return new ArrayList<>();});RetryUtil.getRetryStatus("執(zhí)行結(jié)束: ", retry);CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行結(jié)束：", circuitBreaker);return result.get();} }AOP式的調(diào)用方法
首先在連接器方法上使用@Retry(name="",fallbackMethod="")注解，其中name是要使用的重試器實(shí)例的名稱，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;} }要求和熔斷器一致，但是需要注意同時(shí)注解重試組件和熔斷器的話，是按照第二種方案來的，即每一次請求都會(huì)被熔斷器記錄。
之后直接調(diào)用方法：
?
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("執(zhí)行結(jié)束：", retryRegistry.retry("backendA"));CircuitBreakerUtil.getCircuitBreakerStatus("執(zhí)行結(jié)束：", circuitBreakerRegistry.circuitBreaker("backendA"));return result;} }使用測試
異常A和B在application.yml文件中設(shè)定為都需要重試，因?yàn)槭褂玫谝环N方案，所以不需要將CallNotPermittedException設(shè)定在重試組件的白名單中，同時(shí)為了測試重試過程中的異常是否會(huì)被熔斷器記錄，將異常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使用另一個(gè)遠(yuǎn)程服務(wù)接口的實(shí)現(xiàn)，將num%4==2的情況返回null，測試根據(jù)返回結(jié)果進(jìn)行重試的功能：
?
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("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫的正常查詢r(jià)eturn repository.findAll();} }同時(shí)添加一個(gè)類自定義哪些返回值需要重試，設(shè)定為返回值為空就進(jìn)行重試，這樣num % 4 == 2時(shí)就可以測試不拋異常，根據(jù)返回結(jié)果進(jìn)行重試了：
?
public class RetryOnResultPredicate implements Predicate {@Overridepublic boolean test(Object o) {return o == null ? true : false;} }使用CircuitBreakerServiceImpl中的AOP或者程序式調(diào)用方法進(jìn)行單元測試，循環(huán)調(diào)用10次：
?
public class CircuitBreakerServiceImplTest{@Autowiredprivate CircuitBreakerServiceImpl circuitService;@Testpublic void circuitBreakerRetryTest() {for (int i=0; i<10; i++){// circuitService.circuitBreakerRetryAOP();circuitService.circuitBreakerRetryNotAOP();}} }看一下運(yùn)行結(jié)果：
?
count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: state= metrics[ failedRetryNum=0, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=1 ] 執(zhí)行結(jié)束：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 1 重試：第1次 count的值 = 2 重試：第2次 count的值 = 3 服務(wù)調(diào)用失敗：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，需要重試 執(zhí)行結(jié)束: state= metrics[ failedRetryNum=1, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=1 ] 執(zhí)行結(jié)束：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=2, failedCalls=1, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ]這部分結(jié)果可以看出來，重試最大次數(shù)設(shè)置為3結(jié)果其實(shí)只重試了2次，服務(wù)共執(zhí)行了3次，重試3次后熔斷器只記錄了1次。而且返回值為null時(shí)也確實(shí)進(jìn)行重試了。
?
服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: state= metrics[ failedRetryNum=2, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=3 ] 執(zhí)行結(jié)束：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=0 ] 已經(jīng)被熔斷，停止重試 執(zhí)行結(jié)束: state= metrics[ failedRetryNum=2, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=3 ] 執(zhí)行結(jié)束：state=OPEN , metrics[ failureRate=40.0, bufferedCalls=5, failedCalls=2, successCalls=3, maxBufferCalls=5, notPermittedCalls=1 ]當(dāng)熔斷之后不會(huì)再進(jìn)行重試。
接下來我修改一下調(diào)用服務(wù)的實(shí)現(xiàn)：
?
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("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫的正常查詢r(jià)eturn repository.findAll();} }將num%4==2變成異常B，num%4==3變成返回null，看一下最后一次重試返回值為null屬于重試成功還是重試失敗。
運(yùn)行結(jié)果如下：
?
count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: state= metrics[ failedRetryNum=0, failedNotRetryNum=0, successfulRetryNum=0, successfulNotyRetryNum=1 ] 執(zhí)行結(jié)束：state=CLOSED , metrics[ failureRate=-1.0, bufferedCalls=1, failedCalls=0, successCalls=1, maxBufferCalls=5, notPermittedCalls=0 ] count的值 = 1 重試：第1次 count的值 = 2 重試：第2次 count的值 = 3 服務(wù)調(diào)用成功：2019-07-09T19:17:35.836+08:00[Asia/Shanghai]: Retry 'backendA' recorded a successful retry attempt. Number of retry attempts: '3', Last exception was: 'com.example.resilience4j.exceptions.BusinessBException: 異常B，需要重試'.如上可知如果最后一次重試不拋出異常就算作重試成功，不管結(jié)果是否需要繼續(xù)重試。
Bulkhead
簡介
Resilence4j的Bulkhead提供兩種實(shí)現(xiàn)，一種是基于信號量的，另一種是基于有等待隊(duì)列的固定大小的線程池的，由于基于信號量的Bulkhead能很好地在多線程和I/O模型下工作，所以選擇介紹基于信號量的Bulkhead的使用。
可配置參數(shù)
| maxConcurrentCalls | 25 | 可允許的最大并發(fā)線程數(shù) |
| maxWaitDuration | 0 | 嘗試進(jìn)入飽和艙壁時(shí)應(yīng)阻止線程的最大時(shí)間 |
測試前準(zhǔn)備
pom.xml
不需要引入新的依賴，已經(jīng)集成在resilience4j-spring-boot中了
application.yml配置
?
resilience4j:bulkhead:configs:default:maxConcurrentCalls: 10maxWaitDuration: 1000instances:backendA:baseConfig: defaultmaxConcurrentCalls: 3backendB:baseConfig: defaultmaxWaitDuration: 100和CircuitBreaker差不多，都是可以通過繼承覆蓋配置設(shè)定實(shí)例的。
用于監(jiān)控Bulkhead狀態(tài)及事件的工具類
同樣為了監(jiān)控Bulkhead組件，寫一個(gè)工具類：
?
@Log4j2 public class BulkhdadUtil {/*** @Description: 獲取bulkhead的狀態(tài)*/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: 監(jiān)聽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()));} }調(diào)用方法
還是以之前查詢用戶列表的服務(wù)為例。Bulkhead支持AOP和程序式兩種方式的調(diào)用。
程序式的調(diào)用方法
調(diào)用方法都類似，裝飾方法之后用Try.of().recover()來執(zhí)行：
?
public class BulkheadServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate BulkheadRegistry bulkheadRegistry;public List<User> bulkheadNotAOP(){// 通過注冊器獲得Bulkhead實(shí)例Bulkhead bulkhead = bulkheadRegistry.bulkhead("backendA");BulkhdadUtil.getBulkheadStatus("開始執(zhí)行前: ", bulkhead);// 通過Try.of().recover()調(diào)用裝飾后的服務(wù)Try<List<User>> result = Try.of(Bulkhead.decorateCheckedSupplier(bulkhead, remoteServiceConnector::process)).recover(BulkheadFullException.class, throwable -> {log.info("服務(wù)失敗: " + throwable.getLocalizedMessage());return new ArrayList();});BulkhdadUtil.getBulkheadStatus("執(zhí)行結(jié)束: ", bulkhead);return result.get();} }AOP式的調(diào)用方法
首先在連接器方法上使用@Bulkhead(name="", fallbackMethod="", type="")注解，其中name是要使用的Bulkhead實(shí)例的名稱，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("服務(wù)失敗: " + e.getLocalizedMessage());return new ArrayList();} }如果Retry、CircuitBreaker、Bulkhead同時(shí)注解在方法上，默認(rèn)的順序是Retry>CircuitBreaker>Bulkhead，即先控制并發(fā)再熔斷最后重試，之后直接調(diào)用方法：
?
public class BulkheadServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate BulkheadRegistry bulkheadRegistry;public List<User> bulkheadAOP() throws TimeoutException, InterruptedException {List<User> result = remoteServiceConnector.process();BulkheadUtil.getBulkheadStatus("執(zhí)行結(jié)束：", bulkheadRegistry.retry("backendA"));return result;} }使用測試
在application.yml文件中將backenA線程數(shù)限制為1，便于觀察，最大等待時(shí)間為1s，超過1s的會(huì)走降級方法：
?
instances:backendA:baseConfig: defaultmaxConcurrentCalls: 1使用另一個(gè)遠(yuǎn)程服務(wù)接口的實(shí)現(xiàn)，不拋出異常，當(dāng)做正常服務(wù)進(jìn)行：
?
public class RemoteServiceImpl implements RemoteService {private static AtomicInteger count = new AtomicInteger(0);public List<User> process() {int num = count.getAndIncrement();log.info("count的值 = " + num);log.info("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫正常查詢r(jià)eturn repository.findAll();} }用線程池調(diào)5個(gè)線程去請求服務(wù)：
?
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());}} }看一下運(yùn)行結(jié)果：
?
開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' permitted a call. count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 開始執(zhí)行前: , 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. 服務(wù)失敗: Bulkhead 'backendA' is full and does not permit further calls 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務(wù)失敗: Bulkhead 'backendA' is full and does not permit further calls 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務(wù)失敗: Bulkhead 'backendA' is full and does not permit further calls 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] 服務(wù)失敗: Bulkhead 'backendA' is full and does not permit further calls 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' has finished a call. 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ]由上可以看出，5個(gè)請求只有一個(gè)進(jìn)入，其余觸發(fā)rejected事件，然后自動(dòng)進(jìn)入降級方法。接下來我們把等待時(shí)間稍微加長一些：
?
instances:backendA:baseConfig: defaultmaxConcurrentCalls: 1maxWaitDuration: 5000再運(yùn)行一次：
?
開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] 開始執(zhí)行前: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' permitted a call. count的值 = 0 服務(wù)正常運(yùn)行，獲取用戶列表 Bulkhead 'backendA' permitted a call. count的值 = 1 Bulkhead 'backendA' has finished a call. 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' has finished a call. 執(zhí)行結(jié)束: , metrics[ availableConcurrentCalls=1, maxAllowedConcurrentCalls=1 ] Bulkhead 'backendA' permitted a call.前面的線程沒有馬上被拒絕，而是等待了一段時(shí)間再執(zhí)行。
RateLimiter
簡介
高頻控制是可以限制服務(wù)調(diào)用頻率，Resilience4j的RateLimiter可以對頻率進(jìn)行納秒級別的控制，在每一個(gè)周期刷新可以調(diào)用的次數(shù)，還可以設(shè)定線程等待權(quán)限的時(shí)間。
可配置參數(shù)
| timeoutDuration | 5[s] | 線程等待權(quán)限的默認(rèn)等待時(shí)間 |
| limitRefreshPeriod | 500[ns] | 權(quán)限刷新的時(shí)間，每個(gè)周期結(jié)束后，RateLimiter將會(huì)把權(quán)限計(jì)數(shù)設(shè)置為limitForPeriod的值 |
| limiteForPeriod | 50 | 一個(gè)限制刷新期間的可用權(quán)限數(shù) |
測試前準(zhǔn)備
pom.xml
不需要引入新的依賴，已經(jīng)集成在resilience4j-spring-boot中了
application.yml配置
?
resilience4j:ratelimiter:configs:default:limitForPeriod: 5limitRefreshPeriod: 1stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1backendB:baseConfig: defaulttimeoutDuration: 0s用于監(jiān)控RateLimiter狀態(tài)及事件的工具類
同樣為了監(jiān)控RateLimiter組件，寫一個(gè)工具類：
?
@Log4j2 public class RateLimiterUtil {/*** @Description: 獲取rateLimiter的狀態(tài)*/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: 監(jiān)聽rateLimiter事件*/public static void addRateLimiterListener(RateLimiter rateLimiter){rateLimiter.getEventPublisher().onSuccess(event -> log.info(event.toString())).onFailure(event -> log.info(event.toString()));} }調(diào)用方法
還是以之前查詢用戶列表的服務(wù)為例。RateLimiter支持AOP和程序式兩種方式的調(diào)用。
程序式的調(diào)用方法
調(diào)用方法都類似，裝飾方法之后用Try.of().recover()來執(zhí)行：
?
public class RateLimiterServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate RateLimiterRegistry rateLimiterRegistry;public List<User> ratelimiterNotAOP(){// 通過注冊器獲得RateLimiter實(shí)例RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("backendA");RateLimiterUtil.getRateLimiterStatus("開始執(zhí)行前: ", rateLimiter);// 通過Try.of().recover()調(diào)用裝飾后的服務(wù)Try<List<User>> result = Try.of(Bulkhead.decorateCheckedSupplier(rateLimiter, remoteServiceConnector::process)).recover(BulkheadFullException.class, throwable -> {log.info("服務(wù)失敗: " + throwable.getLocalizedMessage());return new ArrayList();});RateLimiterUtil.getRateLimiterStatus("執(zhí)行結(jié)束: ", rateLimiter);return result.get();} }AOP式的調(diào)用方法
首先在連接器方法上使用@RateLimiter(name="", fallbackMethod="")注解，其中name是要使用的RateLimiter實(shí)例的名稱，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("服務(wù)失敗: " + e.getLocalizedMessage());return new ArrayList();} }如果Retry、CircuitBreaker、Bulkhead、RateLimiter同時(shí)注解在方法上，默認(rèn)的順序是Retry>CircuitBreaker>RateLimiter>Bulkhead，即先控制并發(fā)再限流然后熔斷最后重試
接下來直接調(diào)用方法：
?
public class RateLimiterServiceImpl {@Autowiredprivate RemoteServiceConnector remoteServiceConnector;@Autowiredprivate RateLimiterRegistry rateLimiterRegistry;public List<User> rateLimiterAOP() throws TimeoutException, InterruptedException {List<User> result = remoteServiceConnector.process();BulkheadUtil.getBulkheadStatus("執(zhí)行結(jié)束：", rateLimiterRegistry.retry("backendA"));return result;} }使用測試
在application.yml文件中將backenA設(shè)定為20s只能處理1個(gè)請求，為便于觀察，刷新時(shí)間設(shè)定為20s，等待時(shí)間設(shè)定為5s：
?
configs:default:limitForPeriod: 5limitRefreshPeriod: 20stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1使用另一個(gè)遠(yuǎn)程服務(wù)接口的實(shí)現(xiàn)，不拋出異常，當(dāng)做正常服務(wù)進(jìn)行，為了讓結(jié)果明顯一些，讓方法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("服務(wù)正常運(yùn)行，獲取用戶列表");// 模擬數(shù)據(jù)庫正常查詢r(jià)eturn repository.findAll();} }用線程池調(diào)5個(gè)線程去請求服務(wù)：
?
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());}} }看一下測試結(jié)果：
?
開始執(zhí)行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , 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]} 服務(wù)失敗: RateLimiter 'backendA' does not permit further calls 服務(wù)失敗: RateLimiter 'backendA' does not permit further calls 執(zhí)行結(jié)束: , metrics[ availablePermissions=0, numberOfWaitingThreads=1 ] 執(zhí)行結(jié)束: , metrics[ availablePermissions=0, numberOfWaitingThreads=1 ] RateLimiterEvent{type=FAILED_ACQUIRE, rateLimiterName='backendA', creationTime=2019-07-10T17:06:20.745+08:00[Asia/Shanghai]} 服務(wù)正常運(yùn)行，獲取用戶列表 服務(wù)失敗: RateLimiter 'backendA' does not permit further calls 執(zhí)行結(jié)束: , metrics[ availablePermissions=0, numberOfWaitingThreads=0 ] 服務(wù)失敗: RateLimiter 'backendA' does not permit further calls 執(zhí)行結(jié)束: , metrics[ availablePermissions=0, numberOfWaitingThreads=0 ] 執(zhí)行結(jié)束: , metrics[ availablePermissions=1, numberOfWaitingThreads=0 ]只有一個(gè)服務(wù)調(diào)用成功，其他都執(zhí)行失敗了。現(xiàn)在我們把刷新時(shí)間調(diào)成1s：
?
configs:default:limitForPeriod: 5limitRefreshPeriod: 1stimeoutDuration: 5sinstances:backendA:baseConfig: defaultlimitForPeriod: 1重新執(zhí)行，結(jié)果如下：
?
開始執(zhí)行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 開始執(zhí)行前: , 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 服務(wù)正常運(yùn)行，獲取用戶列表 服務(wù)正常運(yùn)行，獲取用戶列表 服務(wù)正常運(yùn)行，獲取用戶列表 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 執(zhí)行結(jié)束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 執(zhí)行結(jié)束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ]執(zhí)行結(jié)束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ] 服務(wù)正常運(yùn)行，獲取用戶列表 執(zhí)行結(jié)束: , metrics[ availablePermissions=2, numberOfWaitingThreads=0 ]可以看出，幾個(gè)服務(wù)都被放入并正常執(zhí)行了，即使上個(gè)服務(wù)還沒完成，依然可以放入，只與時(shí)間有關(guān)，而與線程無關(guān)。
?
總結(jié)
以上是生活随笔為你收集整理的Resilience4j-轻量级熔断框架的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： JSR 303 - Bean Valid
- 下一篇： Zookeeper集群脑裂问题