SpringCloud（11）：Hystrix请求合并

1 简介

如图，多个客户端发送请求调用(消费者)项目中的findOne方法，这时候在这个项目中的线程池中会发申请与请求数量相同的线程数，对EurekaServiceProvider(服务提供者)的getUserById方法发起调用，每个线程都要调用一次，在高并发的场景下，这样势必会对服务提供者项目产生巨大的压力。

请求合并就是将单个请求合并成一个请求，去调用服务提供者，从而降低服务提供者负载的，一种应对高并发的解决办法

2 请求合并的原理

NetFlix在Hystrix为我们提供了应对高并发的解决方案----请求合并，如下图

通过请求合并器设置延迟时间，将时间内的，多个请求单个的对象的方法中的参数（id）取出来，拼成符合服务提供者的多个对象返回接口（getUsersByIds方法）的参数，指定调用这个接口（getUsersByIds方法），返回的对象List再通过一个方法（mapResponseToRequests方法），按照请求的次序将结果对象对应的装到Request对应的Response中返回结果。

3 请求合并适用的场景

在服务提供者提供了返回单个对象和多个对象的查询接口，并且单个对象的查询并发数很高，服务提供者负载较高的时候，我们就可以使用请求合并来降低服务提供者的负载

4 请求合并带来的问题

问题：即然请求合并这么好，我们是否就可以将所有返回单个结果的方法都用上请求合并呢？答案自然是否定的！

原因：

1. 我们为这个请求人为的设置了延迟时间，这样在并发不高的接口上使用请求缓存，会降低响应速度

2. 实现请求合并比较复杂

5 实现方式

修改OrderService，添加下面方法

    @HystrixCollapser(batchMethod = "findAll", collapserProperties = {@HystrixProperty(name = "timerDelayInMilliseconds", value = "300")})public Future<String> findOne(Integer id) {System.out.println("被合并的请求");return null;}@HystrixCommandpublic List<String> findAll(List<Integer> ids) {System.out.println("合并的请求");String url = "http://study-user/getUserAll?ids={1}";return restTemplate.getForObject(url, List.class, StringUtils.join(ids, ","));}

在OrderController进行测试

    @RequestMapping("/findTest")public String findTest() throws ExecutionException, InterruptedException {HystrixRequestContext context = HystrixRequestContext.initializeContext();Future<String> f1 = orderService.findOne(1);Future<String> f2 = orderService.findOne(2);Thread.sleep(150);Future<String> f3 = orderService.findOne(3);String res = f1.get() + " , " + f2.get() + " , " + f3.get();context.close();return res;}

测试接口

查看日志输出

注解讲解：

以下面代码为例

@HystrixCommand(groupKey = "productStockOpLog", commandKey = "addProductStockOpLog",
fallbackMethod = "addProductStockOpLogFallback",
commandProperties = {
@HystrixProperty(name =
"execution.isolation.thread.timeoutInMilliseconds", value = "400"),//指定多久超时，单位毫秒。超
时进fallback
@HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value
= "10"),//判断熔断的最少请求数，默认是10；只有在一个统计窗口内处理的请求数量达到这个阈值，才会进行熔断与否
的判断
@HystrixProperty(name = "circuitBreaker.errorThresholdPercentage",
value = "10"),//判断熔断的阈值，默认值50，表示在一个统计窗口内有50%的请求处理失败，会触发熔断
}
)
public void addProductStockOpLog(Long sku_id, Object old_value, Object new_value) throws
Exception {
if (new_value != null && !new_value.equals(old_value)) {
doAddOpLog(null, null, sku_id, null, ProductOpType.PRODUCT_STOCK, old_value != null
? String.valueOf(old_value) : null, String.valueOf(new_value), 0, "C端", null);
}
}
public void addProductStockOpLogFallback(Long sku_id, Object old_value, Object new_value)
throws Exception {
LOGGER.warn("发送商品库存变更消息失败,进入Fallback,skuId:{},oldValue:{},newValue:{}",
sku_id, old_value, new_value);
}

@HystrixCommand(groupKey="UserGroup", commandKey = "GetUserByIdCommand"，
commandProperties = {
@HystrixProperty(name =
"execution.isolation.thread.timeoutInMilliseconds", value = "100"),//指定多久超时，单位毫秒。超
时进fallback
@HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value
= "10"),//判断熔断的最少请求数，默认是10；只有在一个统计窗口内处理的请求数量达到这个阈值，才会进行熔断与否
的判断
@HystrixProperty(name = "circuitBreaker.errorThresholdPercentage",
value = "10"),//判断熔断的阈值，默认值50，表示在一个统计窗口内有50%的请求处理失败，会触发熔断
},
threadPoolProperties = {
@HystrixProperty(name = "coreSize", value = "30"),
@HystrixProperty(name = "maxQueueSize", value = "101"),
@HystrixProperty(name = "keepAliveTimeMinutes", value = "2"),
@HystrixProperty(name = "queueSizeRejectionThreshold", value = "15"),
@HystrixProperty(name = "metrics.rollingStats.numBuckets", value =
"12"),
@HystrixProperty(name = "metrics.rollingStats.timeInMilliseconds",
value = "1440")
})

超时时间（默认1000ms，单位：ms）

（1）hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds

在调用方配置，被该调用方的所有方法的超时时间都是该值，优先级低于下边的指定配置

（2）hystrix.command.HystrixCommandKey.execution.isolation.thread.timeoutInMilliseconds

在调用方配置，被该调用方的指定方法（HystrixCommandKey方法名）的超时时间是该值

线程池核心线程数 hystrix.threadpool.default.coreSize（默认为10）

Queue

(1)hystrix.threadpool.default.maxQueueSize（最大排队长度。默认-1，使用SynchronousQueue。其他值则使用 LinkedBlockingQueue。如果要从-1换成其他值则需重启，即该值不能动态调整，若要动态调整，需要使用到下边这个配置）

(2)hystrix.threadpool.default.queueSizeRejectionThreshold（排队线程数量阈值，默认为5，达到时拒绝，如果配置了该选项，队列的大小是该队列）

注意：如果maxQueueSize=-1的话，则该选项不起作用

断路器

（1）hystrix.command.default.circuitBreaker.requestVolumeThreshold（当在配置时间窗口内达到此数量的失败后，进行短路。默认20个）

For example, if the value is 20, then if only 19 requests are received in the rolling

window (say a window of 10 seconds) the circuit will not trip open even if all 19 failed.

简言之，10s内请求失败数量达到20个，断路器开。

（2）hystrix.command.default.circuitBreaker.sleepWindowInMilliseconds（短路多久以后开始尝试是否恢复，默认5s）

（3）hystrix.command.default.circuitBreaker.errorThresholdPercentage（出错百分比阈值，当达到此阈值后，开始短路。默认50%）

fallback

hystrix.command.default.fallback.isolation.semaphore.maxConcurrentRequests（调用线程允许请求HystrixCommand.GetFallback()的最大数量，默认10。超出时将会有异常抛出，注意：该项配置对于THREAD隔离模式也起作用）