Spring Cloud 中斷路器 Circuit Breaker 的應(yīng)用

作者：Spring源碼解讀 2021-12-09 08:57:46

SpringCloud Circuit breaker(斷路器)提供了跨不同斷路器實(shí)現(xiàn)的抽象。它提供了在應(yīng)用程序中使用的一致API，允許開(kāi)發(fā)人員選擇最適合應(yīng)用程序需要的斷路器實(shí)現(xiàn)。

[[439483]]

環(huán)境：Springboot2.3.12.RELEASE +

cloud-netflix-hystrix2.2.10.RELEASE

簡(jiǎn)介

支持的斷路器類(lèi)型：

Netfix Hystrix
Resilience4J
Sentinel
Spring Retry

核心概念

要在代碼中創(chuàng)建斷路器(circuit breaker)，可以使用斷路器工廠API。當(dāng)您在類(lèi)路徑中包含Spring Cloud Circuit Breaker starter時(shí)，將自動(dòng)創(chuàng)建一個(gè)實(shí)現(xiàn)此API的bean。下面給出了使用此API的一個(gè)非常簡(jiǎn)單的示例：

@Service 
public static class DemoService { 
  private RestTemplate rest; 
  private CircuitBreakerFactory cbFactory; 
 
  public DemoService(RestTemplate rest, CircuitBreakerFactory cbFactory) { 
    this.rest = rest; 
    this.cbFactory = cbFactory; 
  } 
 
  public String slow() { 
    // 通過(guò)默認(rèn)的CircuitBreakerFactory工廠創(chuàng)建一個(gè)指定id（名稱(chēng)）的斷路器 
    // run方法是實(shí)際執(zhí)行你的業(yè)務(wù)方法，第二個(gè)參數(shù)throwable 是當(dāng)發(fā)生異常或者是執(zhí)行超時(shí) 
    // 執(zhí)行的回退（降級(jí)）處理 
    return cbFactory.create("slow").run(() -> rest.getForObject("/slow", String.class), throwable -> "fallback"); 
  } 
}

項(xiàng)目配置

通過(guò)引入下面不同依賴(lài)來(lái)確定使用具體的那個(gè)斷路器：

Hystrix - org.springframework.cloud:spring-cloud-starter-netflix-hystrix
Resilience4J - org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j
Reactive Resilience4J - org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j
Spring Retry - org.springframework.cloud:spring-cloud-starter-circuitbreaker-spring-retry
Sentinal - org.springframework.cloud:spring-cloud-starter-circuitbreaker-sentinal

以上5種斷路器是不同的實(shí)現(xiàn)方式，根據(jù)需要引入即可。

示例

這里以Hystrix為例來(lái)使用

引入依賴(lài)

<dependency> 
  <groupId>org.springframework.cloud</groupId> 
  <artifactId>spring-cloud-starter-netflix-hystrix</artifactId> 
  <version>2.2.10.RELEASE</version> 
</dependency>

定義具有熔斷功能的服務(wù)

@Service 
public class DemoService { 
 
  private RestTemplate rest; 
  // 注入系統(tǒng)默認(rèn)的實(shí)現(xiàn) 
  private CircuitBreakerFactory cbFactory; 
 
  public DemoService(RestTemplate rest, CircuitBreakerFactory cbFactory) { 
    this.rest = rest; 
    this.cbFactory = cbFactory; 
  } 
 
  public String slow() { 
    // 使用系統(tǒng)默認(rèn)的實(shí)現(xiàn)創(chuàng)建斷路器進(jìn)行業(yè)務(wù)的處理 
    return cbFactory.create("slow").run(() -> rest.getForObject("http://localhost:8080/demos/slow", String.class), throwable -> "fallback"); 
  } 
 
  public String slow2() { 
    // 使用自定義的斷路器工廠進(jìn)行業(yè)務(wù)的處理 
    return cbf().create("demo-slow").run(() -> rest.getForObject("http://localhost:8080/demos/slow", String.class), throwable -> "fallback"); 
  } 
 
  // 可以將這個(gè)定義為Bean來(lái)覆蓋系統(tǒng)默認(rèn)的實(shí)現(xiàn)，在系統(tǒng)默認(rèn)的實(shí)現(xiàn)上有條件限定 
  private CircuitBreakerFactory<HystrixCommand.Setter, HystrixCircuitBreakerFactory.HystrixConfigBuilder> cbf() { 
    HystrixCircuitBreakerFactory cbf = new HystrixCircuitBreakerFactory() ; 
    // 配置線程池 
    HystrixThreadPoolProperties.Setter threadPoolProperties = HystrixThreadPoolProperties.Setter() ; 
    threadPoolProperties.withCoreSize(5) 
      .withKeepAliveTimeMinutes(5) 
      .withMaxQueueSize(Integer.MAX_VALUE) 
      .withQueueSizeRejectionThreshold(1000) ; 
    // 配置默認(rèn)的執(zhí)行行為屬性 
    HystrixCommandProperties.Setter commandProperties = HystrixCommandProperties.Setter() ; 
    commandProperties.withCircuitBreakerEnabled(true) 
    // 當(dāng)請(qǐng)求超過(guò)了3s那么斷路器就會(huì)工作進(jìn)行回退（降級(jí)處理），執(zhí)行上面run方法中的第二個(gè)參數(shù) 
      .withExecutionTimeoutInMilliseconds(3000) 
      .withRequestCacheEnabled(true) 
      // 隔離策略有兩種THREAD，SEMAPHORE 
      // THREAD: 避免線程被阻塞 
      // SEMAPHORE: 適合高并發(fā)限流處理；因?yàn)榫€程池的方式一般不會(huì)創(chuàng)建過(guò)多的線程 
      // 線程是有限的，在高并發(fā)情況下是沒(méi)法滿(mǎn)足響應(yīng)處理的。 
      .withExecutionIsolationStrategy(ExecutionIsolationStrategy.THREAD); 
         
    // 將其加入到集合中，為不同的服務(wù)創(chuàng)建不同的配置 
    cbf.configure(builder -> { 
      builder.commandProperties(commandProperties).groupName("demo") ; 
    }, "demo-slow"); 
         
    // 當(dāng)默認(rèn)的id不存在時(shí)使用這默認(rèn)的配置 
    cbf.configureDefault(id -> { 
      HystrixCommand.Setter setter = HystrixCommand.Setter 
        .withGroupKey(HystrixCommandGroupKey.Factory.asKey("demo")) // 服務(wù)分組，大的模塊 
        .andCommandKey(HystrixCommandKey.Factory.asKey("demo-slow")) // 服務(wù)標(biāo)識(shí)（具體服務(wù)分組中的某一個(gè)子的服務(wù)），子模塊 
        .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("demo-pools")) // 線程池名稱(chēng) 
        .andThreadPoolPropertiesDefaults(threadPoolProperties) // 線程池相關(guān)配置 
        .andCommandPropertiesDefaults(commandProperties) ; // 執(zhí)行時(shí)相關(guān)屬性配置  
      return setter ; 
    }); 
    return cbf ; 
  } 
 
}

Controller接口

@RestController 
@RequestMapping("/demos") 
public class DemoController { 
     
  @Resource 
  private DemoService demoService ; 
     
  @GetMapping("/index") 
  public Object index() { 
    return demoService.slow2() ; 
  } 
     
  @GetMapping("/slow") 
  public Object slow() { 
    try { 
      TimeUnit.SECONDS.sleep(5) ; 
    } catch (InterruptedException e) { 
      e.printStackTrace(); 
    } 
    return "slow" ; 
  } 
     
}

原理

CircuitBreakerFactory#create方法創(chuàng)建了CircuitBreaker實(shí)例。

根據(jù)當(dāng)前的CLASSPATH我們使用的是Hystrix，那么這里使用的工廠就是：

HystrixCircuitBreakerFactory類(lèi)

public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory<HystrixCommand.Setter, HystrixCircuitBreakerFactory.HystrixConfigBuilder> {

泛型參數(shù)：Setter就是用來(lái)配置Hystrix相關(guān)配置信息的(這里主要用來(lái)CommandKey與Setter進(jìn)行綁定)，HystrixConfigBuilder用來(lái)構(gòu)建 HystrixCommand.Setter對(duì)象。

當(dāng)執(zhí)行HystrixCircuitBreakerFactory#configure方法時(shí)：

public abstract class AbstractCircuitBreakerFactory<CONF, CONFB extends ConfigBuilder<CONF>> { 
  private final ConcurrentHashMap<String, CONF> configurations = new ConcurrentHashMap<>(); 
  public void configure(Consumer<CONFB> consumer, String... ids) { 
    for (String id : ids) { 
      // 構(gòu)建一個(gè)Builder對(duì)象 
      CONFB builder = configBuilder(id); 
      // 這里通過(guò)builder(HystrixConfigBuilder)對(duì)象來(lái)應(yīng)用Consumer中編寫(xiě)的配置信息 
      consumer.accept(builder); 
      // 構(gòu)建HystrixCommand.Setter 對(duì)象 
      CONF conf = builder.build(); 
      // 最后將通過(guò)id 與 Setter對(duì)象綁定key=value存入Map集合中 
      getConfigurations().put(id, conf); 
    } 
  } 
  // 該方法在子類(lèi)HystrixCircuitBreakerFactory中實(shí)現(xiàn) 
  protected abstract CONFB configBuilder(String id); 
}

斷路器具體的子類(lèi)實(shí)現(xiàn)

HystrixCircuitBreakerFactory

// 子類(lèi)繼承的父類(lèi)中的泛型：第一個(gè)泛型參數(shù)：需要構(gòu)建什么樣的一個(gè)配置，第二個(gè)泛型參數(shù)：通過(guò)誰(shuí)來(lái)構(gòu)建第一個(gè)泛型參數(shù)配置 
public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory<HystrixCommand.Setter, HystrixCircuitBreakerFactory.HystrixConfigBuilder> { 
  public HystrixConfigBuilder configBuilder(String id) { 
    return new HystrixConfigBuilder(id); 
  } 
  public static class HystrixConfigBuilder extends AbstractHystrixConfigBuilder<HystrixCommand.Setter> { 
    public HystrixConfigBuilder(String id) { 
      super(id); 
    } 
    // 從這里也看出來(lái)最終Builder就是用來(lái)構(gòu)建Setter對(duì)象用 
    @Override 
    public HystrixCommand.Setter build() { 
      return HystrixCommand.Setter.withGroupKey(getGroupKey()) 
        .andCommandKey(getCommandKey()) 
        .andCommandPropertiesDefaults(getCommandPropertiesSetter()); 
    } 
  } 
}

斷路器工廠有了，接下來(lái)就是通過(guò)工廠創(chuàng)建具體的斷路器對(duì)象了。

通過(guò)上面的代碼執(zhí)行cbf().create("demo-slow")方法時(shí)執(zhí)行了什么?

public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory<HystrixCommand.Setter, HystrixCircuitBreakerFactory.HystrixConfigBuilder> { 
  private Function<String, HystrixCommand.Setter> defaultConfiguration = id -> HystrixCommand.Setter 
    .withGroupKey(HystrixCommandGroupKey.Factory.asKey(getClass().getSimpleName())) 
    .andCommandKey(HystrixCommandKey.Factory.asKey(id)); 
  public HystrixCircuitBreaker create(String id) { 
    // 通過(guò)上面分析最終所有的Hystrix的Setter會(huì)與id綁定存入一個(gè)Map中 
    // 這里computeIfAbsent方法先從集合中通過(guò)id獲取，如果獲取不到則將第二個(gè)參數(shù)存入集合中返回 
    HystrixCommand.Setter setter = getConfigurations().computeIfAbsent(id, defaultConfiguration); 
    return new HystrixCircuitBreaker(setter); 
  } 
}

上面創(chuàng)建的是HystrixCircuitBreaker斷路器，當(dāng)執(zhí)行run方法時(shí)：

public class HystrixCircuitBreaker implements CircuitBreaker { 
  private HystrixCommand.Setter setter; 
  public HystrixCircuitBreaker(HystrixCommand.Setter setter) { 
    this.setter = setter; 
  } 
  @Override 
  public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) { 
    // 最終執(zhí)行的就是Hystrix的核心 HystrixCommand對(duì)象 
    HystrixCommand<T> command = new HystrixCommand<T>(setter) { 
      @Override 
      protected T run() throws Exception { 
        return toRun.get(); 
      } 
      @Override 
      protected T getFallback() { 
        return fallback.apply(getExecutionException()); 
      } 
    }; 
    return command.execute(); 
  } 
}