Spring?Boot?Reactor?整合?Resilience4j詳析

更新時(shí)間：2022年09月22日 14:25:24 作者：六七十三

這篇文章主要介紹了Spring?Boot?Reactor整合Resilience4j詳析，文章通過引入pom包展開詳細(xì)介紹，具有一定的參考價(jià)值，感興趣的小伙伴可以參考一下

1 引入 pom 包

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-all</artifactId>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
</dependency>

2 配置說明

2.1 限流 ratelimiter

兩個(gè)限流配置：backendA 1s 中最多允許 10 次請(qǐng)求；

backendB 每 500ms 最多允許 6 次請(qǐng)求。

resilience4j.ratelimiter:
  instances:
    backendA:
      limitForPeriod: 10
      limitRefreshPeriod: 1s
      timeoutDuration: 10ms
      registerHealthIndicator: true
      eventConsumerBufferSize: 100
    backendB:
      limitForPeriod: 6
      limitRefreshPeriod: 500ms
      timeoutDuration: 3s

配置屬性	默認(rèn)值	描述
timeoutDuration	5【s】	一個(gè)線程等待許可的默認(rèn)等待時(shí)間
limitRefreshPeriod	500【ns】	限制刷新的周期。在每個(gè)周期之后，速率限制器將其權(quán)限計(jì)數(shù)設(shè)置回 limitForPeriod 值
limitForPeriod	50	一個(gè) limitRefreshPeriod （周期）允許訪問的數(shù)量（許可數(shù)量）

2.2 重試 retry

注意指定需要重試的異常，不是所有的異常重試都有效。比如 DB 相關(guān)校驗(yàn)異常，如唯一約束等，重試也不會(huì)成功的。

重試配置：

resilience4j.retry:
  instances:
    backendA:
      maxAttempts: 3
      waitDuration: 10s
      enableExponentialBackoff: true
      exponentialBackoffMultiplier: 2
      retryExceptions:
        - org.springframework.web.client.HttpServerErrorException
        - java.io.IOException

    backendB:
      maxAttempts: 3
      waitDuration: 10s
      retryExceptions:
        - org.springframework.web.client.HttpServerErrorException
        - java.io.IOException

配置屬性	默認(rèn)值	描述
maxAttempts	3	最大重試次數(shù)（包括第一次）
waitDuration	500【ms】	兩次重試之間的等待間隔
intervalFunction	numOfAttempts -> waitDuration	修改失敗后等待間隔的函數(shù)。默認(rèn)情況下，等待時(shí)間是個(gè)常量。
retryOnResultPredicate	result->false	配置一個(gè)判斷結(jié)果是否應(yīng)該重試的 predicate 函數(shù)。如果結(jié)果應(yīng)該重試，Predicate 必須返回 true，否則它必須返回 false。
retryExceptionPredicate	throwable -> true	和 retryOnResultPredicate 類似，如果要重試，Predicate 必須返回true，否則返回 false。
retryExceptions	空	需要重試的異常類型列表
ignoreExceptions	空	不需要重試的異常類型列表
failAfterMaxAttempts	false	當(dāng)重試達(dá)到配置的 maxAttempts 并且結(jié)果仍未通過 retryOnResultPredicate 時(shí)啟用或禁用拋出 MaxRetriesExceededException 的布爾值
intervalBiFunction	(numOfAttempts, Either<throwable, result>) -> waitDuration	根據(jù) maxAttempts 和結(jié)果或異常修改失敗后等待間隔時(shí)間的函數(shù)。與 intervalFunction 一起使用時(shí)會(huì)拋出 IllegalStateException。

2.3 超時(shí) TimeLimiter

超時(shí)配置：

resilience4j.timelimiter:
  instances:
    backendA:
      timeoutDuration: 2s
      cancelRunningFuture: true
    backendB:
      timeoutDuration: 1s
      cancelRunningFuture: false

超時(shí)配置比較簡(jiǎn)單，主要是配置 timeoutDuration 也就是超時(shí)的時(shí)間。

cancelRunningFuture 的意思是：是否應(yīng)該在運(yùn)行的 Future 調(diào)用 cancel 去掉調(diào)用。

2.4 斷路器 circuitbreaker

斷路器有幾種狀態(tài)：關(guān)閉、打開、半開。注意：打開，意味著不能訪問，會(huì)迅速失敗。

CircuitBreaker 使用滑動(dòng)窗口來存儲(chǔ)和匯總調(diào)用結(jié)果。您可以在基于計(jì)數(shù)的滑動(dòng)窗口和基于時(shí)間的滑動(dòng)窗口之間進(jìn)行選擇?；谟?jì)數(shù)的滑動(dòng)窗口聚合最后 N 次調(diào)用的結(jié)果。基于時(shí)間的滑動(dòng)窗口聚合了最后 N 秒的調(diào)用結(jié)果。

斷路器配置：

resilience4j.circuitbreaker:
  instances:
    backendA:
      // 健康指標(biāo)參數(shù)，非斷路器屬性
      registerHealthIndicator: true
      slidingWindowSize: 100

配置屬性	默認(rèn)值	描述
slidingWindowSize	100	記錄斷路器關(guān)閉狀態(tài)下（可以訪問的情況下）的調(diào)用的滑動(dòng)窗口大小
failureRateThreshold	50（百分比）	當(dāng)失敗比例超過 failureRateThreshold 的時(shí)候，斷路器會(huì)打開，并開始短路呼叫
slowCallDurationThreshold	60000【ms】	請(qǐng)求被定義為慢請(qǐng)求的閾值
slowCallRateThreshold	100（百分比）	慢請(qǐng)求百分比大于等于該值時(shí)，打開斷路器開關(guān)
permittedNumberOfCalls	10	半開狀態(tài)下允許通過的請(qǐng)求數(shù)
maxWaitDurationInHalfOpenState	0	配置最大等待持續(xù)時(shí)間，該持續(xù)時(shí)間控制斷路器在切換到打開之前可以保持在半開狀態(tài)的最長(zhǎng)時(shí)間。

值 0 表示斷路器將在 HalfOpen 狀態(tài)下無限等待，直到所有允許的調(diào)用都已完成。

2.5 壁倉 bulkhead

resilience4j 提供了兩種實(shí)現(xiàn)壁倉的方法：

SemaphoreBulkhead 使用 Semaphore 實(shí)現(xiàn)
FixedThreadPoolBulkhead 使用有界隊(duì)列和固定線程池實(shí)現(xiàn)

resilience4j.bulkhead:
  instances:
    backendA:
      maxConcurrentCalls: 10
    backendB:
      maxWaitDuration: 10ms
      maxConcurrentCalls: 20

resilience4j.thread-pool-bulkhead:
  instances:
    backendC:
      maxThreadPoolSize: 1
      coreThreadPoolSize: 1
      queueCapacity: 1

2.5.1 SemaphoreBulkhead

配置屬性	默認(rèn)值	描述
maxConcurrentCalls	25	允許的并發(fā)執(zhí)行的數(shù)量
maxWaitDuration	0	嘗試進(jìn)入飽和隔板時(shí)線程應(yīng)被阻止的最長(zhǎng)時(shí)間

2.5.2 FixedThreadPoolBulkhead

配置屬性	默認(rèn)值	描述
maxThreadPoolSize	Runtime.getRuntime().availableProcessors()	線程池最大線程個(gè)數(shù)
coreThreadPoolSize	Runtime.getRuntime().availableProcessors()-1	線程池核心線程個(gè)數(shù)
queueCapacity	100	線程池隊(duì)列容量
keepAliveDuration	20【ms】	線程數(shù)超過核心線程數(shù)之后，空余線程在終止之前等待的最長(zhǎng)時(shí)間

3 使用

3.1 配置

在 application.yml 文件中添加以下 resilience4j 配置：

resilience4j.circuitbreaker:
  instances:
    backendA:
      registerHealthIndicator: true
      slidingWindowSize: 100

resilience4j.retry:
  instances:
    backendA:
      maxAttempts: 3
      waitDuration: 10s
      enableExponentialBackoff: true
      exponentialBackoffMultiplier: 2
      retryExceptions:
        - org.springframework.web.client.HttpServerErrorException
        - java.io.IOException

    backendB:
      maxAttempts: 3
      waitDuration: 10s
      retryExceptions:
        - org.springframework.web.client.HttpServerErrorException
        - java.io.IOException


resilience4j.bulkhead:
  instances:
    backendA:
      maxConcurrentCalls: 10
    backendB:
      maxWaitDuration: 10ms
      maxConcurrentCalls: 20

resilience4j.thread-pool-bulkhead:
  instances:
    backendC:
      maxThreadPoolSize: 1
      coreThreadPoolSize: 1
      queueCapacity: 1

resilience4j.ratelimiter:
  instances:
    backendA:
      limitForPeriod: 10
      limitRefreshPeriod: 1s
      timeoutDuration: 10ms
      registerHealthIndicator: true
      eventConsumerBufferSize: 100
    backendB:
      limitForPeriod: 6
      limitRefreshPeriod: 500ms
      timeoutDuration: 3s

resilience4j.timelimiter:
  instances:
    backendA:
      timeoutDuration: 2s
      cancelRunningFuture: true
    backendB:
      timeoutDuration: 1s
      cancelRunningFuture: false

3.2 使用注解實(shí)現(xiàn)

直接在需要限流的方法上增加注解@RateLimiter 實(shí)現(xiàn)限流；增加注解@Retry 實(shí)現(xiàn)重試；增加注解 @CircuitBreaker 熔斷；增加注解 @Bulkhead 實(shí)現(xiàn)壁倉。name 屬性中分別填寫限流器、重試、熔斷、壁倉組件的名字。

@Bulkhead(name = "backendA")
@CircuitBreaker(name = "backendA")
@Retry(name = "backendA")
@RateLimiter(name = "backendA")
public Mono<List<User>> list() {
  long startTime = System.currentTimeMillis();
  return Mono.fromSupplier(() -> {
        return userRepository.findAll();
      }).doOnError(e -> {
        // 打印異常日志&增加監(jiān)控（自行處理）
        logger.error("list.user.error, e", e);
      })
      .doFinally(e -> {
        // 耗時(shí) & 整體健康
        logger.info("list.user.time={}, ", System.currentTimeMillis() - startTime);
      });
}
  
@Bulkhead(name = "backendA")
@CircuitBreaker(name = "backendA")//最多支持10個(gè)并發(fā)量
@Retry(name = "backendA")//使用 backendA 重試器，如果拋出 IOException 會(huì)重試三次。
@RateLimiter(name = "backendA")// 限流 10 Qps
public Mono<Boolean> save(User user) {
  long startTime = System.currentTimeMillis();
  return Mono.fromSupplier(() -> {
        return userRepository.save(user) != null;
      })
      .doOnError(e -> {
        // 打印異常日志&增加監(jiān)控（自行處理）
        logger.error("save.user.error, user={}, e", user, e);
      })
      .doFinally(e -> {
        // 耗時(shí) & 整體健康
        logger.info("save.user.time={}, user={}", user, System.currentTimeMillis() - startTime);
      });
}

注意：以上所有組件，都支持自定義。

到此這篇關(guān)于Spring Boot Reactor 整合 Resilience4j詳析的文章就介紹到這了,更多相關(guān)Spring Boot Reactor 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: