hystrix，request collapser，请求合并

hystrix，request collapser，请求合并

多个商品，需要发送多次网络请求，调用多次接口，才能拿到结果

可以使用HystrixCollapser将多个HystrixCommand合并到一起，多个command放在一个command里面去执行，发送一次网络请求，就拉取到多条数据

用请求合并技术，将多个请求合并起来，可以减少高并发访问下需要使用的线程数量以及网络连接数量，这都是hystrix自动进行的

其实对于高并发的访问来说，是可以提升性能的

请求合并有很多种级别

（1）global context，tomcat所有调用线程，对一个依赖服务的任何一个command调用都可以被合并在一起，hystrix就传递一个HystrixRequestContext

（2）user request context，tomcat内某一个调用线程，将某一个tomcat线程对某个依赖服务的多个command调用合并在一起

（3）object modeling，基于对象的请求合并，如果有几百个对象，遍历后依次调用每个对象的某个方法，可能导致发起几百次网络请求，基于hystrix可以自动将对多个对象模型的调用合并到一起

请求合并技术的开销有多大

使用请求合并技术的开销就是导致延迟大幅度增加，因为需要一定的时间将多个请求合并起来

发送过来10个请求，每个请求本来大概是2ms可以返回，要把10个请求合并在一个command内，统一一起执行，先后等待一下，5ms

所以说，要考量一下，使用请求合并技术是否合适，如果一个请求本来耗费的时间就比较长，那么进行请求合并，增加一些延迟影响并不大请求合并技术，不是针对那种访问延时特别低的请求的，比如说你的访问延时本身就比较高，20ms，10个请求合并在一起，25ms，这种情况下就还好

好处在哪里，大幅度削减你的线程池的资源耗费，线程池，10个线程，一秒钟可以执行10个请求，合并在一起，1个线程执行10个请求，10个线程就可以执行100个请求

增加你的吞吐量

减少你对后端服务访问时的网络资源的开销，10个请求，10个command，10次网络请求的开销，1次网络请求的开销了

每个请求就2ms，batch，8~10ms，延迟增加了4~5倍

每个请求本来就30ms~50ms，batch，35ms~55ms，延迟增加不太明显

将多个command请求合并到一个command中执行

请求合并时，可以设置一个batch size，以及elapsed time（控制什么时候触发合并后的command执行）

有两种合并模式，一种是request scope，另一种是global scope，默认是rquest scope，在collapser构造的时候指定scope模式

request scope的batch收集是建立在一个request context内的，而global scope的batch收集是横跨多个request context的

所以对于global context来说，必须确保能在一个command内处理多个requeset context的请求

在netflix，是只用request scope请求合并的，因为默认是用唯一一个request context包含所有的command，所以要做合并，肯定就是request scope

一般请求合并技术，对于那种访问同一个资源的command，但是参数不同，是很有效的

批量查询，HystrixObservableCommand，HystrixCommand+request cache，都是每个商品发起一次网络请求

一个批量的商品过来以后，我们还是多个command的方式去执行，request collapser+request cache，相同的商品还是就查询一次，不同的商品合并到一起通过一个网络请求得到结果

import java.util.Collection;import java.util.List;import com.alibaba.fastjson.JSONArray;import com.netflix.hystrix.HystrixCollapser;import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;import com.netflix.hystrix.HystrixCommandKey;import com.roncoo.eshop.cache.ha.com.roncoo.eshop.cache.ha.model.ProductInfo;public class GetProductInfosCollapser extends HystrixCollapser, ProductInfo, Long> { private Long productId; public GetProductInfosCollapser(Long productId) { this.productId = productId; } @Override public Long getRequestArgument() { return productId; } @Override protected HystrixCommand> createCommand( Collection> requests) { StringBuilder paramsBuilder = new StringBuilder(""); for(CollapsedRequest request : requests) { paramsBuilder.append(request.getArgument()).append(","); } String params = paramsBuilder.toString(); params = params.substring(0, params.length() - 1); System.out.println("createCommand方法执行，params=" + params); return new BatchCommand(requests); } @Override protected void mapResponseToRequests( List batchResponse, Collection> requests) { int count = 0; for(CollapsedRequest request : requests) { request.setResponse(batchResponse.get(count++)); } } @Override protected String getCacheKey() { return "product_info_" + productId; } private static final class BatchCommand extends HystrixCommand> { public final Collection> requests; public BatchCommand(Collection> requests) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductInfoService")) .andCommandKey(HystrixCommandKey.Factory.asKey("GetProductInfosCollapserBatchCommand"))); this.requests = requests; } @Override protected List run() throws Exception { // 将一个批次内的商品id给拼接在了一起 StringBuilder paramsBuilder = new StringBuilder(""); for(CollapsedRequest request : requests) { paramsBuilder.append(request.getArgument()).append(","); } String params = paramsBuilder.toString(); params = params.substring(0, params.length() - 1); // 在这里，我们可以做到什么呢，将多个商品id合并在一个batch内，直接发送一次网络请求，获取到所有的结果 String url = "+ params; String response = HttpClientUtils.sendGetRequest(url); List productInfos = JSONArray.parseArray(response, ProductInfo.class); for(ProductInfo productInfo : productInfos) { System.out.println("BatchCommand内部，productInfo=" + productInfo); } return productInfos; } }}

import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.ResponseBody;import com.alibaba.fastjson.JSONArray;import com.alibaba.fastjson.JSONObject;public class ProductController { @RequestMapping("/getProductInfos") @ResponseBody public String getProductInfos(String productIds){ System.out.println("getProductInfos接口，接收到一次请求，productIds=" + productIds); JSONArray jsonArray=new JSONArray(); for(String productId :productIds.split(",")){ String json = "{\"id\": " + productId + ", \"name\": \"iphone7手机\", \"price\": 5599, \"pictureList\":\"a.jpg,b.jpg\", \"specification\": \"iphone7的规格\", \"service\": \"iphone7的售后服务\", \"color\": \"红色,白色,黑色\", \"size\": \"5.5\", \"shopId\": 1, \"modifiedTime\": \"2017-01-01 12:00:00\", \"cityId\": 1, \"brandId\": 1}"; jsonArray.add(JSONObject.parseObject(json)); } return jsonArray.toJSONString(); }}

List> futures =new ArrayList>(); for(String productId:productIds.split(",")){ GetProductInfosCollapser getProductInfosCollapser = new GetProductInfosCollapser(Long.valueOf(productId)); futures.add(getProductInfosCollapser.queue()); } try { for(Future future : futures) { System.out.println("CacheController的结果：" + future.get()); } } catch (Exception e) { e.printStackTrace(); }

（1）maxRequestsInBatch

控制一个Batch中最多允许多少个request被合并，然后才会触发一个batch的执行

默认值是无限大，就是不依靠这个数量来触发执行，而是依靠时间

HystrixCollapserProperties.Setter() .withMaxRequestsInBatch(int value)

（2）timerDelayInMilliseconds

控制一个batch创建之后，多长时间以后就自动触发batch的执行，默认是10毫秒

HystrixCollapserProperties.Setter() .withTimerDelayInMilliseconds(int value)

super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("GetProductInfosCollapser")) .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter() .withMaxRequestsInBatch(100) .withTimerDelayInMilliseconds(20)));

fail-fast，就是不给fallback降级逻辑，HystrixCommand.run()，直接报错，直接会把这个报错抛出来，给你的tomcat调用线程

fail-silent，给一个fallback降级逻辑，如果HystrixCommand.run()，报错了，会走fallback降级，直接返回一个空值，HystrixCommand，就给一个null

HystrixObservableCommand，Observable.empty()

很少会用fail-fast模式，比较常用的可能还是fail-silent，特别常用，既然都到了fallback里面，肯定要做点降级的事情

public class FailureModeCommand extends HystrixCommand{private boolean failure; public FailureModeCommand(boolean failure) { super(HystrixCommandGroupKey.Factory.asKey("FailureModeGroup")); this.failure = failure; } @Override protected Boolean run() throws Exception { if(failure) { throw new Exception(); } return true; } @Override protected Boolean getFallback() { return false; }}

stubbed fallback，残缺的降级

用请求中的部分数据拼装成结果，然后再填充一些默认值，返回

比如说你发起了一个请求，然后请求中可能本身就附带了一些信息，如果主请求失败了，走到降级逻辑

在降级逻辑里面，可以将这个请求中的数据，以及部分本地缓存有的数据拼装在一起，再给数据填充一些简单的默认值

然后尽可能将自己有的数据返回到请求方

stubbed，残缺了，比如说应该查询到一个商品信息，里面包含20个字段

请求参数搂出来一两个字段，从本地的少量缓存中比如说，可以搂出来那么两三个字段，最终的话返回的字段可能就五六个，其他的字段都是填充的默认值

数据有残缺。

多级降级

先降一级，尝试用一个备用方案去执行，如果备用方案失败了，再用最后下一个备用方案去执行

command嵌套command

尝试从备用服务器接口去拉取结果

mysql数据库访问报错，降级，去redis中获取数据

如果说redis又挂了，然后就去从本地ehcache缓存中获取数据

hystrix command fallback语义，很容易就可以实现多级降级的策略

command，fallback，又套了一个command，第二个command其实是第一级降级策略

第二个command的fallback是第二级降级策略

第一级降级策略，可以是

storm，我们之前做storm这块，第一级降级，一般是搞一个storm的备用机房，部署了一套一模一样的拓扑，如果主机房中的storm拓扑挂掉了，备用机房的storm拓扑定顶上

如果备用机房的storm拓扑也挂了

第二级降级，可能就降级成用mysql/hbase/redis/es

第三级降级，离线批处理去做，hdfs+spark，每个小时执行一次数据统计，去降级

主流程，访问的商品服务，是从主机房去访问的，服务，如果主机房的服务出现了故障，机房断电，机房的网络负载过高，机器硬件出了故障

第一级降级策略，去访问备用机房的服务

第二级降级策略，用stubbed fallback降级策略，比较常用的，返回一些残缺的数据回去

private static class FirstLevelFallbackCommand extends HystrixCommand { private Long productId; public FirstLevelFallbackCommand(Long productId) { // 第一级的降级策略，因为这个command是运行在fallback中的 // 所以至关重要的一点是，在做多级降级的时候，要将降级command的线程池单独做一个出来 // 如果主流程的command都失败了，可能线程池都已经被占满了 // 降级command必须用自己的独立的线程池 super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductInfoService")) .andCommandKey(HystrixCommandKey.Factory.asKey("FirstLevelFallbackCommand")) .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("FirstLevelFallbackPool")) ); this.productId = productId; } @Override protected ProductInfo run() throws Exception { // 这里，因为是第一级降级的策略，所以说呢，其实是要从备用机房的机器去调用接口 // 但是，我们这里没有所谓的备用机房，所以说还是调用同一个服务来模拟 if(productId.equals(-2L)) { throw new Exception(); } String url = "+ productId; String response = HttpClientUtils.sendGetRequest(url); return JSONObject.parseObject(response, ProductInfo.class); } @Override protected ProductInfo getFallback() { // 第二级降级策略，第一级降级策略，都失败了 ProductInfo productInfo = new ProductInfo(); // 从请求参数中获取到的唯一条数据 productInfo.setId(productId); // 从本地缓存中获取一些数据 productInfo.setBrandId(BrandCache.getBrandId(productId)); productInfo.setBrandName(BrandCache.getBrandName(productInfo.getBrandId())); productInfo.setCityId(LocationCache.getCityId(productId)); productInfo.setCityName(LocationCache.getCityName(productInfo.getCityId())); // 手动填充一些默认的数据 productInfo.setColor("默认颜色"); productInfo.setModifiedTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())); productInfo.setName("默认商品"); productInfo.setPictureList("default.jpg"); productInfo.setPrice(0.0); productInfo.setService("默认售后服务"); productInfo.setShopId(-1L); productInfo.setSize("默认大小"); productInfo.setSpecification("默认规格"); return productInfo; } }

手动降级

你写一个command，在这个command它的主流程中，根据一个标识位，判断要执行哪个流程

可以执行主流程，command，也可以执行一个备用降级的command

一般来说，都是去执行一个主流程的command，如果说你现在知道有问题了，希望能够手动降级的话，动态给服务发送个请求

在请求中修改标识位，自动就让command以后都直接过来执行备用command

3个command，套在最外面的command，是用semaphore信号量做限流和资源隔离的，因为这个command不用去care timeout的问题，嵌套调用的command会自己去管理timeout超时的。

商品服务接口的手动降级的方案

主流程还是去走GetProductInfoCommand，手动降级的方案，比如说是从某一个数据源，自己去简单的获取一些数据，尝试封装一下返回

手动降级的策略，就比较low了，调用别人的接口去获取数据的，业务逻辑的封装

主流程有问题，那么可能你就需要立即自己写一些逻辑发布上去，从mysql数据库的表中获取一些数据去返回，手动调整一下降级标识，做一下手动降级

import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;import com.netflix.hystrix.HystrixCommandKey;import com.netflix.hystrix.HystrixCommandProperties;import com.netflix.hystrix.HystrixCommandProperties.ExecutionIsolationStrategy;import com.roncoo.eshop.cache.degrade.IsDegrade;import com.roncoo.eshop.cache.ha.model.ProductInfo;public class GetProductInfoFacadeCommand extends HystrixCommand { private Long productId; public GetProductInfoFacadeCommand(Long productId) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductInfoService")) .andCommandKey(HystrixCommandKey.Factory.asKey("GetProductInfoFacadeCommand")) .andCommandPropertiesDefaults(HystrixCommandProperties.Setter() .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE) .withExecutionIsolationSemaphoreMaxConcurrentRequests(15))); this.productId = productId; } @Override protected ProductInfo run() throws Exception { if(!IsDegrade.isDegrade()) { return new GetProductInfoCommand(productId).execute(); } else { return new GetProductInfoFromMySQLCommand(productId).execute(); } } @Override protected ProductInfo getFallback() { return new ProductInfo(); }}

public class IsDegrade { private static boolean degrade = false; public static boolean isDegrade() { return degrade; } public static void setDegrade(boolean degrade) { IsDegrade.degrade = degrade; }}

public class ManualDegradeTest { public static void main(String[] args) throws Exception { GetProductInfoFacadeCommand getProductInfoFacadeCommand1 = new GetProductInfoFacadeCommand(1L); System.out.println(getProductInfoFacadeCommand1.execute()); IsDegrade.setDegrade(true); GetProductInfoFacadeCommand getProductInfoFacadeCommand2 = new GetProductInfoFacadeCommand(1L); System.out.println(getProductInfoFacadeCommand2.execute()); } }

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。