Java：Resilience4j CircuitBreaker入门指南

「爱情、让人受尽委屈。」 2023-07-04 08:43 59阅读 0赞

原文：[https://resilience4j.readme.io/docs/circuitbreaker][https_resilience4j.readme.io_docs_circuitbreaker]

**目录**

介绍

基于计数的滑动窗口

基于时间的滑动窗口

失败率和慢调用率阈值

创建CircuitBreakerRegistry

创建和配置CircuitBreaker

装配和执行函数式接口

消费发出的RegistryEvent

消费发出的CircuitBreakerEvent

--------------------

### 介绍 ###

CircuitBreaker是由一个有限状态机实现的，其中包含三个一般性状态：CLOSED, OPEN, HALF\_OPEN（关闭、打开、半开）和两个特定状态：DISABLED, FORCED\_OPEN（禁用、强开）。

![20200209172238473.jpg][]  
CircuitBreaker使用滑动窗口（sliding window）存储和统计调用的结果。你能在基于计数的滑动窗口（count-based sliding window）和基于时间的滑动窗口（time-based sliding window）中作选择。基于计数的滑动窗口统计最近N次调用的结果。基于时间的滑动窗口统计最近N秒的调用结果。

### 基于计数的滑动窗口 ###

基于计数的滑动窗口是由包含N个测量单元的循环数组实现的。若计数窗口的大小为10，循环数组总会有10个测量单元。滑动窗口会增量更新总统计值。当新的调用结果被记录后，总统计值就会被更新。当最老的测量单元被排除，这个测量单元就会从总统计值中减掉，占用的桶就会被重置。（排除后减掉（Subtract-on-Evict））

获取快照的时间恒定为 O(1)，因为快照是已统计好的，并独立于窗口大小。这种实现的空间需求（内存消耗）是O(n)。

### 基于时间的滑动窗口 ###

基于时间的滑动窗口是由N个部分统计单元（桶）的循环数组实现的。  
若滑动窗口大小为10秒，循环数组含有10个部分统计单元（桶）。每个桶统计一个指定秒内的所有调用的结果（部分统计）。循环数组的头部桶存储当前秒内所有调用的结果。其它部分统计单元存储以前秒内的调用结果。  
滑动窗口不会单独存储所有调用结果，但会增量更新部分统计单元（桶）中的值和总统计值。  
当新的调用结果被记录后，总统计值会增量更新。当最老的桶被排除，这个桶中的部分统计值会从总统计值中减掉，这个桶也会被重置。（排除后减掉（Subtract-on-Evict））

获取快照的时间恒定为 O(1)，因为快照是已统计好的，并独立于窗口大小。因为调用结果不会各自单独存储，所以这种实现的空间需求（内存消耗）接近O(n)。是需要创建N个部分统计单元和一个总统计单元。  
为了给失败调用次数、慢调用次数和总调用次数计数，部分统计单元是由3个integer组成。并且一个long字段存储总调用时长。

### 失败率和慢调用率阈值 ###

当失败率等于或大于阈值时，CircuitBreaker的状态由CLOSED转为OPEN。比如，当大于50%被记录的调用失败了。

所有的异常默认都会记为失败。你可以定义一个可以当作失败的异常列表。除此以外的其它异常，除非被忽略，不然都会被当作成功。异常也可以被忽略，这样它既不被当作失败也不被当作成功。

当慢调用的比例等于或大于阈值时，CircuitBreaker的状态由CLOSED转为OPEN。比如，当大于50%被记录的调用花的时间大于5秒。这将有助于在外部系统变得无响应阄，降低它的负载。

只有当记录到的调用次数达到一定的最小数量时，失败率和慢调用率才能被计算出来。比如，若要求调用的最小次数是10，那最少必须记录到10次调用，失败率才能被计算出来。若只评估了9次调用，哪怕这9次调用都失败，CircuitBreaker也不会打开。

当CircuitBreaker状态为OPEN时，调用将被拒绝，并将异常CallNotPermittedException返回给调用者。当经过一定的等待时间后，CircuitBreaker状态会从OPEN变为HALF\_OPEN，并允许一定数量的调用去执行，以查看后端是仍然不可用还是已再次可用。除非所有被允许的调用都完成，不然其后的调用都会被拒绝并返回异常CallNotPermittedException。

若失败率和慢调用率仍大于等于配置的阈值，状态会改回OPEN。若失败率和慢调用率低于阈值，状态会改回CLOSED。

CircuitBreaker还支持两种特殊的状态：DISABLED（总是允许访问）, FORCED\_OPEN（总是拒绝访问）。处于这两种状态时，没有CircuitBreaker事件（除了状态转换）会产生，也不会记录任何度量值。从这两种状态退出的唯一办法就是触发状态转换或重置CircuitBreaker。

CircuitBreaker是线程安全的：

*  CircuitBreaker的状态存放在AtomicReference中。
 *  CircuitBreaker使用原子操作更新状态，不会有负作用。
 *  记录调用和从时间窗口中读取快照都是同步处理的。

这意味着保证了原子性，在同一时刻只有一个线程可以更新状态或滑动窗口。

但CircuitBreaker不会同步函数调用。也就是说函数调用本身不是临界区的一个部分。否则CircuitBreaker会带来巨大的性能下降，造成性能瓶颈。慢函数调用会对整个系统的性能/呑吐量造成巨大的负面影响。

若20个并发线程请求执行一个函数，并且CircuitBreaker状态为关闭，所有线程都会被允许执行这个函数。即使滑动窗口的大小为15，并不意味着只能有15个调用并发执行。如果你想限制并发线程数，请使用Bulkhead。你可将Bulkhead与CircuitBreaker一起使用。

1个线程和例子

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70][]

3个线程的例子

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 1][]

### 创建CircuitBreakerRegistry ###

Resilience4j会在内存中带一个CircuitBreakerRegistry，这个CircuitBreakerRegistry基于ConcurrentHashMap，提供了线程安全性和原子性保证。你能使用CircuitBreakerRegistry管理（创建和获取）CircuitBreaker实例。可以使用一个全局默认CircuitBreakerConfig为所有CircuitBreaker实例创建一个CircuitBreakerRegistry。

CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();

### 创建和配置CircuitBreaker ###

你可以创建自己定制的全局CircuitBreakerConfig。为了创建一个全局的CircuitBreakerConfig，可以使用CircuitBreakerConfig构建器。可以使用构建器配置如下属性：

<table> 
 <tbody> 
  <tr> 
   <td>配置属性</td> 
   <td>默认值</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>failureRateThreshold</td> 
   <td>50</td> 
   <td>以百分率形式配置失败率阈值。失败率大于等于阈值时，CircuitBreaker转变为打开状态，并使调用短路。</td> 
  </tr> 
  <tr> 
   <td>slowCallRateThreshold</td> 
   <td>100</td> 
   <td>以百分率形式配置慢调用率阈值。当调用执行的时长大于slowCallDurationThreshold时，CircuitBreaker会认为调用为慢调用。当慢调用占比大于等于此阈值时，CircuitBreaker转变为打开状态，并使调用短路。</td> 
  </tr> 
  <tr> 
   <td>slowCallDurationThreshold</td> 
   <td>60000 [ms]</td> 
   <td>配置调用执行的时长阈值。当超过这个阈值时，调用会被认为是慢调用，并增加慢调用率。</td> 
  </tr> 
  <tr> 
   <td>permittedNumberOfCallsInHalfOpenState</td> 
   <td>10</td> 
   <td>当CircuitBreaker是半开状态时，配置被允许的调用次数。</td> 
  </tr> 
  <tr> 
   <td>slidingWindowType</td> 
   <td> <p>COUNT_BASED</p> <p>/TIME_BASED</p> </td> 
   <td>配置滑动窗口类型。当CircuitBreaker关闭时，这种类型的滑动窗口会记录调用结果。滑动窗口要么是基于计数的，要么是基于时间的。若滑动窗口为COUNT_BASED，则最近slidingWindowSize次的调用会被记录和统计。若滑动窗口为TIME_BASED，则最近slidingWindowSize秒中的调用会被记录和统计。</td> 
  </tr> 
  <tr> 
   <td>slidingWindowSize</td> 
   <td>100</td> 
   <td>配置滑动窗口的大小。当CircuitBreaker关闭后用于记录调用结果。</td> 
  </tr> 
  <tr> 
   <td>minimumNumberOfCalls</td> 
   <td>10</td> 
   <td>配置最小调用次数。在CircuitBreaker计算错误率前，要求（在每滑动窗口周期）用到这个值。例如，若minimumNumberOfCalls是10，为计算失败率，则最小要记录10个调用。若只记录了9个调用，即使9个都失败，CircuitBreaker也不会打开。</td> 
  </tr> 
  <tr> 
   <td>waitDurationInOpenState</td> 
   <td>60000 [ms]</td> 
   <td>CircuitBreaker状态从打开转化为半开时，需要等待的时长。</td> 
  </tr> 
  <tr> 
   <td>automaticTransitionFromOpenToHalfOpenEnabled</td> 
   <td>false</td> 
   <td>如果为true，则CircuitBreaker会自动从打开状态转化为半开状态。不需要另外的调用来触发这种转换。</td> 
  </tr> 
  <tr> 
   <td>recordExceptions</td> 
   <td>empty</td> 
   <td>这个异常列表用来存放被当作失败的异常，这些异常发生时会增加失败率。任何异常，只要不是在ignoreExceptions中被明确忽略的，如果匹配或继承自异常列表中的异常，都会当作失败。若设置了异常列表，不在异常列表中的异常，只要不在ignoreExceptions中存在，都会当作成功。</td> 
  </tr> 
  <tr> 
   <td>ignoreExceptions</td> 
   <td>empty</td> 
   <td>这个异常列表用来存放可忽略的异常，这些异常即不当作成功也不当作失败。任何异常，只要匹配或继承自此异常列表中的异常，都不会当作成功或失败，即使recordExceptions中存在这个异常。</td> 
  </tr> 
  <tr> 
   <td>recordException</td> 
   <td> <p>throwable -&gt; true</p> <p>所有异常默认都当作失败。</p> </td> 
   <td>这是一个定制的Predicate，会评估是否把异常当作失败。除非异常在ignoreExceptions中存在，不然，只有Predicate必须返回true时，才把异常当作失败；只有Predicate必须返回false时，才把异常当作成功。</td> 
  </tr> 
  <tr> 
   <td>ignoreException</td> 
   <td> <p>throwable -&gt; false</p> <p>所有异常默认都不会被忽略。</p> </td> 
   <td>这是一个定制的Predicate，会评估异常是否应该被忽略，而不是把它当作失败或成功。如果应忽略此异常，则Predicate必须返回true。如果把异常当作失败，则Predicate必须返回false。</td> 
  </tr> 
 </tbody> 
</table>

// Create a custom configuration for a CircuitBreaker
    CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
      .failureRateThreshold(50)
      .slowCallRateThreshold(50)
      .waitDurationInOpenState(Duration.ofMillis(1000))
      .slowCallDurationThreshold(Duration.ofSeconds(2))
      .permittedNumberOfCallsInHalfOpenState(3)
      .minimumNumberOfCalls(10)
      .slidingWindowType(SlidingWindowType.TIME_BASED)
      .slidingWindowSize(5)
      .recordException(e -> INTERNAL_SERVER_ERROR
                     .equals(getResponse().getStatus()))
      .recordExceptions(IOException.class, TimeoutException.class)
      .ignoreExceptions(BusinessException.class, OtherBusinessException.class)
      .build();
    
    // Create a CircuitBreakerRegistry with a custom global configuration
    CircuitBreakerRegistry circuitBreakerRegistry 
      CircuitBreakerRegistry.of(circuitBreakerConfig);
    
    // Get or create a CircuitBreaker from the CircuitBreakerRegistry 
    // with the global default configuration
    CircuitBreaker circuitBreakerWithDefaultConfig = 
      circuitBreakerRegistry.circuitBreaker("name1");
    
    // Get or create a CircuitBreaker from the CircuitBreakerRegistry 
    // with a custom configuration
    CircuitBreaker circuitBreakerWithCustomConfig = circuitBreakerRegistry
      .circuitBreaker("name2", circuitBreakerConfig);

可以添加被多个CircuitBreaker实例共享的配置。

CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
      .failureRateThreshold(70)
      .build();
    
    circuitBreakerRegistry.addConfiguration("someSharedConfig", config);
    
    CircuitBreaker circuitBreaker = circuitBreakerRegistry
      .circuitBreaker("name", "someSharedConfig");

也可以重写配置。

CircuitBreakerConfig defaultConfig = circuitBreakerRegistry
       .getDefaultConfig();
    
    CircuitBreakerConfig overwrittenConfig = CircuitBreakerConfig
      .from(defaultConfig)
      .waitDurationInOpenState(Duration.ofSeconds(20))
      .build();

如果不想使用CircuitBreakerRegistry管理CircuitBreaker实例，可以直接创建它。

// Create a custom configuration for a CircuitBreaker
    CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
      .recordExceptions(IOException.class, TimeoutException.class)
      .ignoreExceptions(BusinessException.class, OtherBusinessException.class)
      .build();
    
    CircuitBreaker customCircuitBreaker = CircuitBreaker
      .of("testName", circuitBreakerConfig);

### 装配和执行函数式接口 ###

可以在CircuitBreaker上装配做任意Callable, Supplier, Runnable, Consumer, CheckedRunnable, CheckedSupplier, CheckedConsumer, CompletionStage

可以使用来自Vavr的Try.of(…)、Try.run(…)调用装饰函数。这样就可以使用map, flatMap, filter, recover, andThen将更多的函数串起来。这些串起来的函数只有在CircuitBreaker的状态为CLOSED或HALF\_OPEN时，才能被调用。

在下面的例子中，如果函数调用成功，Try.of(…)返回Success<String>。如果函数抛出异常，Failure<Throwable>被返回，并且不会调用map。

// Given
    CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
    
    // When I decorate my function
    CheckedFunction0<String> decoratedSupplier = CircuitBreaker
            .decorateCheckedSupplier(circuitBreaker, () -> "This can be any method which returns: 'Hello");
    
    // and chain an other function with map
    Try<String> result = Try.of(decoratedSupplier)
                    .map(value -> value + " world'");
    
    // Then the Try Monad returns a Success<String>, if all functions ran successfully.
    assertThat(result.isSuccess()).isTrue();
    assertThat(result.get()).isEqualTo("This can be any method which returns: 'Hello world'");

### 消费发出的RegistryEvent ###

在CircuitBreakerRegistry上注册事件消费者，这样在CircuitBreaker被创建、替代、删除时可以执行一些操作。

CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();
    circuitBreakerRegistry.getEventPublisher()
      .onEntryAdded(entryAddedEvent -> {
        CircuitBreaker addedCircuitBreaker = entryAddedEvent.getAddedEntry();
        LOG.info("CircuitBreaker {} added", addedCircuitBreaker.getName());
      })
      .onEntryRemoved(entryRemovedEvent -> {
        CircuitBreaker removedCircuitBreaker = entryRemovedEvent.getRemovedEntry();
        LOG.info("CircuitBreaker {} removed", removedCircuitBreaker.getName());
      });

### 消费发出的CircuitBreakerEvent ###

CircuitBreakerEvent可以是状态转换、断路器重置、成功的调用、已记录的错误、已忽略的错误等事件。所有的事件都包含事件创建时间、调用处理时长等额外信息。若要消费事件，可以注册一个事件消费者。

circuitBreaker.getEventPublisher()
        .onSuccess(event -> logger.info(...))
        .onError(event -> logger.info(...))
        .onIgnoredError(event -> logger.info(...))
        .onReset(event -> logger.info(...))
        .onStateTransition(event -> logger.info(...));
    // Or if you want to register a consumer listening
    // to all events, you can do:
    circuitBreaker.getEventPublisher()
        .onEvent(event -> logger.info(...));

也可以使用CircularEventConsumer将事件存储在固定容量的环形缓存中。

CircularEventConsumer<CircuitBreakerEvent> ringBuffer = 
      new CircularEventConsumer<>(10);
    circuitBreaker.getEventPublisher().onEvent(ringBuffer);
    List<CircuitBreakerEvent> bufferedEvents = ringBuffer.getBufferedEvents()

可以使用RxJava或RxJava2选配器将EventPublisher转化为Reactive Stream。

[https_resilience4j.readme.io_docs_circuitbreaker]: https://resilience4j.readme.io/docs/circuitbreaker
[20200209172238473.jpg]: https://img-blog.csdnimg.cn/20200209172238473.jpg
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200209171429968.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200209171335241.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp,size_16,color_FFFFFF,t_70