SpringCloud CircuitBreaker斷路器應用詳解

作者：Springboot實戰案例錦集 2023-03-13 08:02:55

在上面我們所看到的API，如：CircuitBreaker，CircuitBreakerFactory這些接口(抽象類)的定義都在Spring Cloud Commons包中定義，而具體的實現我們還需要自己根據情況引入。如Resilience4J或者Sentinel，本篇內容使用Resilience4J。

環境：Springboot2.6.14 + Spring Cloud2021.0.5

概述

Spring Cloud Circuit breaker提供了一個跨不同斷路器實現的抽象。它為你的應用程序提供了一致的API，讓你(開發人員)選擇最適合你應用程序需求的斷路器實現。

Spring Cloud支持以下斷路器實現:

Resilience4J
Sentinel
Spring Retry

核心概念

要在代碼中創建斷路器，可以使用CircuitBreakerFactory API。當你在classpath中包含SpringCloud CircuitBreaker啟動器時，會自動為你創建一個實現了這個API的bean。下面的例子展示了如何使用這個API:?

@Service
public static class DemoControllerService {
  private RestTemplate rest;
  private CircuitBreakerFactory cbFactory;


  public DemoControllerService(RestTemplate rest, CircuitBreakerFactory cbFactory) {
    this.rest = rest;
    this.cbFactory = cbFactory;
  }


  public String slow() {
    return cbFactory.create("slow").run(() -> rest.getForObject("/slow", String.class), throwable -> "fallback");
  }


}

以上是斷路器的基本使用方法。CircuitBreakerFactory.create API創建一個名為CircuitBreaker的類實例。run方法接受一個Supplier和一個Function作為參數。Supplier是要封裝在斷路器中的代碼。該函數是在斷路器跳閘時運行的備用機制也就是會調用run方法的第二個參數Function。

實現原理

自動配置

在上面我們所看到的API，如：CircuitBreaker，CircuitBreakerFactory這些接口(抽象類)的定義都在Spring Cloud Commons包中定義，而具體的實現我們還需要自己根據情況引入。如Resilience4J或者Sentinel，本篇內容使用Resilience4J。

首先，引入Ressilience4J自動配置

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-circuitbreaker-resilience4j</artifactId>
</dependency>

引入該依賴后自動配置CircuitBreakerFactory的實現?

// 可以通過spring.cloud.circuitbreaker.resilience4j.enabled控制是否啟用，默認開啟
@EnableConfigurationProperties(Resilience4JConfigurationProperties.class)
@ConditionalOnProperty(name = { "spring.cloud.circuitbreaker.resilience4j.enabled",
    "spring.cloud.circuitbreaker.resilience4j.blocking.enabled" }, matchIfMissing = true)
public class Resilience4JAutoConfiguration {
  @Autowired(required = false)
  private List<Customizer<Resilience4JCircuitBreakerFactory>> customizers = new ArrayList<>();


  @Bean
  @ConditionalOnMissingBean(CircuitBreakerFactory.class)
  public Resilience4JCircuitBreakerFactory resilience4jCircuitBreakerFactory(
    CircuitBreakerRegistry circuitBreakerRegistry, TimeLimiterRegistry timeLimiterRegistry,
    @Autowired(required = false) Resilience4jBulkheadProvider bulkheadProvider,
    Resilience4JConfigurationProperties resilience4JConfigurationProperties) {
    Resilience4JCircuitBreakerFactory factory = new Resilience4JCircuitBreakerFactory(circuitBreakerRegistry,
      timeLimiterRegistry, bulkheadProvider, resilience4JConfigurationProperties);
    customizers.forEach(customizer -> customizer.customize(factory));
    return factory;
  }  
}

有了上面的工廠類Resilience4JCircuitBreakerFactory后我們就可以通過該工廠進行實際的開發了。

在上面的Bean創建方法參數中有幾個非常重要的核心類：

CircuitBreakerRegistry

該類用來配置注冊創建CircuitBreaker實例。

TimeLimiterRegistry

該類用來注冊配置每一個實例（CircuitBreaker）的時間限制。

Resilience4jBulkheadProvider

該類主要是提供了隔離機制，其內通過BulkheadRegistry及
ThreadPoolBulkheadRegistry注冊配置每一個實例的隔離機制及線程池隔離配置。

核心類Resilience4JCircuitBreakerFactory

在需要進行熔斷降級操作都需用通過該工廠進行創建CircuitBreaker實例?

public class Resilience4JCircuitBreakerFactory {
  private CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();
  private TimeLimiterRegistry timeLimiterRegistry = TimeLimiterRegistry.ofDefaults();
  private ExecutorService executorService = Executors.newCachedThreadPool();
  private Function<String, Resilience4JConfigBuilder.Resilience4JCircuitBreakerConfiguration> defaultConfiguration;


  public Resilience4JCircuitBreakerFactory(CircuitBreakerRegistry circuitBreakerRegistry,
    TimeLimiterRegistry timeLimiterRegistry, Resilience4jBulkheadProvider bulkheadProvider,
    Resilience4JConfigurationProperties resilience4JConfigurationProperties) {
    this.circuitBreakerRegistry = circuitBreakerRegistry;
    this.timeLimiterRegistry = timeLimiterRegistry;
    this.bulkheadProvider = bulkheadProvider;
    // 獲取默認的配置（當沒有為具體的id實例創建配置時，使用該默認配置）
    this.defaultConfiguration = id -> new Resilience4JConfigBuilder(id)
      .circuitBreakerConfig(this.circuitBreakerRegistry.getDefaultConfig())
      .timeLimiterConfig(this.timeLimiterRegistry.getDefaultConfig()).build();
    this.resilience4JConfigurationProperties = resilience4JConfigurationProperties;
  }
}

創建CircuitBreaker實例

public Resilience4JCircuitBreaker create(String id) {
  return create(id, id, this.executorService);
}
private Resilience4JCircuitBreaker create(String id, String groupName,
  ExecutorService circuitBreakerExecutorService) {
  // 根據實例id獲取對應的配置，如果不存在則使用默認的配置
  // 初始是沒有對應id的配置
  Resilience4JConfigBuilder.Resilience4JCircuitBreakerConfiguration config = getConfigurations()
    .computeIfAbsent(id, defaultConfiguration);
  // spring.cloud.circuitbreaker.resilience4j.disableThreadPool 默認false
  if (resilience4JConfigurationProperties.isDisableThreadPool()) {
    return new Resilience4JCircuitBreaker(id, groupName, config.getCircuitBreakerConfig(),
      config.getTimeLimiterConfig(), circuitBreakerRegistry, timeLimiterRegistry,
      Optional.ofNullable(circuitBreakerCustomizers.get(id)), bulkheadProvider);
  } else {
    // 創建CircuitBreaker實例，這里的config.getCircuitBreakerConfig和config.getTimeLimiterConfig
    // 兩個方法返回的默認的配置（首次）
    return new Resilience4JCircuitBreaker(id, groupName, config.getCircuitBreakerConfig(),
      config.getTimeLimiterConfig(), circuitBreakerRegistry, timeLimiterRegistry,
      circuitBreakerExecutorService, Optional.ofNullable(circuitBreakerCustomizers.get(id)),bulkheadProvider);
  }
}

Resilience4JCircuitBreaker實例?

public class Resilience4JCircuitBreaker implements CircuitBreaker {
  public Resilience4JCircuitBreaker(String id, String groupName,
      io.github.resilience4j.circuitbreaker.CircuitBreakerConfig circuitBreakerConfig,
      TimeLimiterConfig timeLimiterConfig, CircuitBreakerRegistry circuitBreakerRegistry,
      TimeLimiterRegistry timeLimiterRegistry, ExecutorService executorService,
      Optional<Customizer<io.github.resilience4j.circuitbreaker.CircuitBreaker>> circuitBreakerCustomizer,
      Resilience4jBulkheadProvider bulkheadProvider) {
    this.id = id;
    this.groupName = groupName;
    this.circuitBreakerConfig = circuitBreakerConfig;
    this.registry = circuitBreakerRegistry;
    this.timeLimiterRegistry = timeLimiterRegistry;
    this.timeLimiterConfig = timeLimiterConfig;
    this.executorService = executorService;
    // ...
  }
  public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) {
    // ...
    // 下面的CircuitBreaker和TimeLimiter都會先從Registry中獲取，不存在則使用默認的
    // 從CircuitBreakerRegistry獲取對應Id實例配置，如果不存在則返回的就是當前circuitBreakerConfig
    io.github.resilience4j.circuitbreaker.CircuitBreaker defaultCircuitBreaker = registry.circuitBreaker(this.id,
        this.circuitBreakerConfig, tags);
    // 如果對應Id實例配置不存在則應用默認的timeLimiterConfig配置
    TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter(id, timeLimiterConfig, tags);
    // ...
  }
}

CircuitBreakerRegistry注冊器

該注冊器將配置文件中配置的實例配置進行注冊并提供默認的配置

@Import({CircuitBreakerConfigurationOnMissingBean.class, FallbackConfigurationOnMissingBean.class})
public class CircuitBreakerAutoConfiguration {}

CircuitBreakerConfigurationOnMissingBean該類注冊CircuitBreakerRegistry

@Configuration
public class CircuitBreakerConfigurationOnMissingBean extends AbstractCircuitBreakerConfigurationOnMissingBean {


  public CircuitBreakerConfigurationOnMissingBean(CircuitBreakerConfigurationProperties circuitBreakerProperties) {
    super(circuitBreakerProperties);
  }
}

屬性配置?

// 相關配置都在該屬性配置中
@ConfigurationProperties(prefix = "resilience4j.circuitbreaker")
public class CircuitBreakerProperties extends CircuitBreakerConfigurationProperties {}

AbstractCircuitBreakerConfigurationOnMissingBean?

@Configuration
@Import({FallbackConfigurationOnMissingBean.class, SpelResolverConfigurationOnMissingBean.class})
public abstract class AbstractCircuitBreakerConfigurationOnMissingBean {
  protected final CircuitBreakerConfiguration circuitBreakerConfiguration;
  protected final CircuitBreakerConfigurationProperties circuitBreakerProperties;
  public AbstractCircuitBreakerConfigurationOnMissingBean(CircuitBreakerConfigurationProperties circuitBreakerProperties) {
    this.circuitBreakerProperties = circuitBreakerProperties;
    this.circuitBreakerConfiguration = new CircuitBreakerConfiguration(circuitBreakerProperties);
  }
  @Bean
  @ConditionalOnMissingBean
  public CircuitBreakerRegistry circuitBreakerRegistry(
      EventConsumerRegistry<CircuitBreakerEvent> eventConsumerRegistry,
      RegistryEventConsumer<CircuitBreaker> circuitBreakerRegistryEventConsumer,
      @Qualifier("compositeCircuitBreakerCustomizer") CompositeCustomizer<CircuitBreakerConfigCustomizer> compositeCircuitBreakerCustomizer) {
    // 創建實例
    return circuitBreakerConfiguration
      .circuitBreakerRegistry(eventConsumerRegistry, circuitBreakerRegistryEventConsumer,compositeCircuitBreakerCustomizer);
  }
}

CircuitBreakerConfiguration?

@Configuration
public class CircuitBreakerConfiguration {
  private final CircuitBreakerConfigurationProperties circuitBreakerProperties;
  public CircuitBreakerConfiguration(CircuitBreakerConfigurationProperties circuitBreakerProperties) {
    this.circuitBreakerProperties = circuitBreakerProperties;
  }
  @Bean
  public CircuitBreakerRegistry circuitBreakerRegistry(
      EventConsumerRegistry<CircuitBreakerEvent> eventConsumerRegistry,
      RegistryEventConsumer<CircuitBreaker> circuitBreakerRegistryEventConsumer,
      @Qualifier("compositeCircuitBreakerCustomizer") CompositeCustomizer<CircuitBreakerConfigCustomizer> compositeCircuitBreakerCustomizer) {
    // 將resilience4j.circuitbreaker.configs所有配置進行注冊
    CircuitBreakerRegistry circuitBreakerRegistry = createCircuitBreakerRegistry(
    circuitBreakerProperties, circuitBreakerRegistryEventConsumer,
    compositeCircuitBreakerCustomizer);
    //...
    // 將resilience4j.circuitbreaker.instances所有配置進行注冊
    initCircuitBreakerRegistry(circuitBreakerRegistry, compositeCircuitBreakerCustomizer);
    return circuitBreakerRegistry;
  }
}

上面的configs和instances所有的配置都會注冊到AbstractRegistry.entryMap中。

獲取實例Id配置

io.github.resilience4j.circuitbreaker.CircuitBreaker defaultCircuitBreaker = registry.circuitBreaker(this.id,this.circuitBreakerConfig, tags);
// InMemoryCircuitBreakerRegistry
public CircuitBreaker circuitBreaker(String name, CircuitBreakerConfig config,io.vavr.collection.Map<String, String> tags) {
  return computeIfAbsent(name, () -> CircuitBreaker.of(name, Objects.requireNonNull(config, CONFIG_MUST_NOT_BE_NULL), getAllTags(tags)));
}
// AbstractRegistry
protected E computeIfAbsent(String name, Supplier<E> supplier) {
  // 從map獲取
  return entryMap.computeIfAbsent(Objects.requireNonNull(name, NAME_MUST_NOT_BE_NULL), k -> {
    E entry = supplier.get();
    eventProcessor.processEvent(new EntryAddedEvent<>(entry));
    return entry;
  });
}

TimeLimiterRegistry注冊器

該配置與CircuitBreakerRegistry的配置類似，這里不贅述

以上就是Resilience4J的實現原理

Resilience4J配置

resilience4j:
  circuitbreaker:
    #configs:
    #  default:
    #    minimumNumberOfCalls: 5 #默認配置
    instances:
      #在Feign中使用的命名
      DemoFeignformat:
        minimumNumberOfCalls: 5
      lkk:
        minimumNumberOfCalls: 2
  timelimiter:
    configs:
      #為任何調用提供默認的配置
      default:
        timeoutDuration: 1s
    instances:
      DemoFeignformat:
        timeoutDuration: 1s
      DemoFeigndate3:
        timeoutDuration: 5s       
      lkk:
        timeoutDuration: 1s

在上面Feign的命名規則如下：

如果開啟了如下配置：?

feign:
  circuitbreaker:
    enabled: true
    alphanumericIds:
      enabled: true

public class FeignClientFactoryBean {
  <T> T getTarget() {
    Feign.Builder builder = feign(context); 
  }
}
public final class FeignCircuitBreaker {
  public static final class Builder extends Feign.Builder {
  public Feign build(final FallbackFactory<?> nullableFallbackFactory) {
    // FeignCircuitBreakerInvocationHandler執行調用
    super.invocationHandlerFactory((target, dispatch) -> new FeignCircuitBreakerInvocationHandler(circuitBreakerFactory, 
      feignClientName, target, dispatch, nullableFallbackFactory,
      circuitBreakerGroupEnabled, circuitBreakerNameResolver));
      return super.build();
    }
  }
}

FeignCircuitBreakerInvocationHandler?

class FeignCircuitBreakerInvocationHandler implements InvocationHandler {
  private final CircuitBreakerNameResolver circuitBreakerNameResolver;
  public Object invoke() {
    String circuitName = circuitBreakerNameResolver.resolveCircuitBreakerName(feignClientName, target, method);
    CircuitBreaker circuitBreaker = circuitBreakerGroupEnabled ? factory.create(circuitName, feignClientName) : factory.create(circuitName);
  }
}
static class AlphanumericCircuitBreakerNameResolver extends DefaultCircuitBreakerNameResolver {


  @Override
  public String resolveCircuitBreakerName(String feignClientName, Target<?> target, Method method) {
    return super.resolveCircuitBreakerName(feignClientName, target, method).replaceAll("[^a-zA-Z0-9]", "");
  }
}
static class DefaultCircuitBreakerNameResolver implements CircuitBreakerNameResolver {


  @Override
  public String resolveCircuitBreakerName(String feignClientName, Target<?> target, Method method) {
    return Feign.configKey(target.type(), method);
  }
}
public abstract class Feign {
  // 開啟了Alphanumeric會將特殊字符全部刪除
  public static String configKey(Class targetType, Method method) {
    StringBuilder builder = new StringBuilder();
    builder.append(targetType.getSimpleName());
    builder.append('#').append(method.getName()).append('(');
    for (Type param : method.getGenericParameterTypes()) {
      param = Types.resolve(targetType, targetType, param);
      builder.append(Types.getRawType(param).getSimpleName()).append(',');
    }
    if (method.getParameterTypes().length > 0) {
      builder.deleteCharAt(builder.length() - 1);
    }
    return builder.append(')').toString();
  }
}

以上就是基于Resilience4J的實現原理及配置實現

責任編輯：武曉燕來源：實戰案例錦集

斷路器應用 API