[Java] Spring Cloud 中断路器 Circuit Breaker的应用

• Netfix Hystrix
  
• Resilience4J
  
• Sentinel
  
• Spring Retry
  

@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() { 
// 通过默认的CircuitBreakerFactory工厂创建一个指定id（名称）的断路器 
// run方法是实际执行你的业务方法，第二个参数throwable 是当发生异常或者是执行超时 
// 执行的回退（降级）处理 
return cbFactory.create("slow").run(() -> rest.getForObject("/slow", String.class), throwable -> "fallback"); 
} 
}

• 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
  

org.springframework.cloud 
spring-cloud-starter-netflix-hystrix 
2.2.10.RELEASE

@Service 
public class DemoService { 

private RestTemplate rest; 
// 注入系统默认的实现 
private CircuitBreakerFactory cbFactory; 

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

public String slow() { 
// 使用系统默认的实现创建断路器进行业务的处理 
return cbFactory.create("slow").run(() -> rest.getForObject("http://localhost:8080/demos/slow", String.class), throwable -> "fallback"); 
} 

public String slow2() { 
// 使用自定义的断路器工厂进行业务的处理 
return cbf().create("demo-slow").run(() -> rest.getForObject("http://localhost:8080/demos/slow", String.class), throwable -> "fallback"); 
} 

// 可以将这个定义为Bean来覆盖系统默认的实现，在系统默认的实现上有条件限定 
private CircuitBreakerFactory cbf() { 
HystrixCircuitBreakerFactory cbf = new HystrixCircuitBreakerFactory() ; 
// 配置线程池 
HystrixThreadPoolProperties.Setter threadPoolProperties = HystrixThreadPoolProperties.Setter() ; 
threadPoolProperties.withCoreSize(5) 
.withKeepAliveTimeMinutes(5) 
.withMaxQueueSize(Integer.MAX_VALUE) 
.withQueueSizeRejectionThreshold(1000) ; 
// 配置默认的执行行为属性 
HystrixCommandProperties.Setter commandProperties = HystrixCommandProperties.Setter() ; 
commandProperties.withCircuitBreakerEnabled(true) 
// 当请求超过了3s那么断路器就会工作进行回退（降级处理），执行上面run方法中的第二个参数 
.withExecutionTimeoutInMilliseconds(3000) 
.withRequestCacheEnabled(true) 
// 隔离策略有两种THREAD，SEMAPHORE 
// THREAD: 避免线程被阻塞 
// SEMAPHORE: 适合高并发限流处理；因为线程池的方式一般不会创建过多的线程 
// 线程是有限的，在高并发情况下是没法满足响应处理的。 
.withExecutionIsolationStrategy(ExecutionIsolationStrategy.THREAD); 

// 将其加入到集合中，为不同的服务创建不同的配置 
cbf.configure(builder -> { 
builder.commandProperties(commandProperties).groupName("demo") ; 
}, "demo-slow"); 

// 当默认的id不存在时使用这默认的配置 
cbf.configureDefault(id -> { 
HystrixCommand.Setter setter = HystrixCommand.Setter 
.withGroupKey(HystrixCommandGroupKey.Factory.asKey("demo")) // 服务分组，大的模块 
.andCommandKey(HystrixCommandKey.Factory.asKey("demo-slow")) // 服务标识（具体服务分组中的某一个子的服务），子模块 
.andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("demo-pools")) // 线程池名称 
.andThreadPoolPropertiesDefaults(threadPoolProperties) // 线程池相关配置 
.andCommandPropertiesDefaults(commandProperties) ; // 执行时相关属性配置 
return setter ; 
}); 
return cbf ; 
} 

}

@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" ; 
} 

}

public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory {

public abstract class AbstractCircuitBreakerFactory

// 子类继承的父类中的泛型：第一个泛型参数：需要构建什么样的一个配置，第二个泛型参数：通过谁来构建第一个泛型参数配置 
public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory { 
public HystrixConfigBuilder configBuilder(String id) { 
return new HystrixConfigBuilder(id); 
} 
public static class HystrixConfigBuilder extends AbstractHystrixConfigBuilder { 
public HystrixConfigBuilder(String id) { 
super(id); 
} 
// 从这里也看出来最终Builder就是用来构建Setter对象用 
@Override 
public HystrixCommand.Setter build() { 
return HystrixCommand.Setter.withGroupKey(getGroupKey()) 
.andCommandKey(getCommandKey()) 
.andCommandPropertiesDefaults(getCommandPropertiesSetter()); 
} 
} 
}

public class HystrixCircuitBreakerFactory extends CircuitBreakerFactory { 
private Function defaultConfiguration = id -> HystrixCommand.Setter 
.withGroupKey(HystrixCommandGroupKey.Factory.asKey(getClass().getSimpleName())) 
.andCommandKey(HystrixCommandKey.Factory.asKey(id)); 
public HystrixCircuitBreaker create(String id) { 
// 通过上面分析最终所有的Hystrix的Setter会与id绑定存入一个Map中 
// 这里computeIfAbsent方法先从集合中通过id获取，如果获取不到则将第二个参数存入集合中返回 
HystrixCommand.Setter setter = getConfigurations().computeIfAbsent(id, defaultConfiguration); 
return new HystrixCircuitBreaker(setter); 
} 
}

public class HystrixCircuitBreaker implements CircuitBreaker { 
private HystrixCommand.Setter setter; 
public HystrixCircuitBreaker(HystrixCommand.Setter setter) { 
this.setter = setter; 
} 
@Override 
public T run(Supplier toRun, Function fallback) { 
// 最终执行的就是Hystrix的核心 HystrixCommand对象 
HystrixCommand command = new HystrixCommand(setter) { 
@Override 
protected T run() throws Exception { 
return toRun.get(); 
} 
@Override 
protected T getFallback() { 
return fallback.apply(getExecutionException()); 
} 
}; 
return command.execute(); 
} 
}