多个商品,需要发送多次网络请求,调用多次接口,才能拿到结果
可以使用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.http.HttpClientUtils; import com.roncoo.eshop.cache.ha.model.ProductInfo; public class GetProductInfosCollapser extends HystrixCollapser<List<ProductInfo>, ProductInfo, Long> { private Long productId; public GetProductInfosCollapser(Long productId) { this.productId = productId; } @Override public Long getRequestArgument() { return productId; } @Override protected HystrixCommand<List<ProductInfo>> createCommand( Collection<com.netflix.hystrix.HystrixCollapser.CollapsedRequest<ProductInfo, Long>> requests) { StringBuilder paramsBuilder = new StringBuilder(""); for(CollapsedRequest<ProductInfo, Long> 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<ProductInfo> batchResponse, Collection<com.netflix.hystrix.HystrixCollapser.CollapsedRequest<ProductInfo, Long>> requests) { int count = 0; for(CollapsedRequest<ProductInfo, Long> request : requests) { request.setResponse(batchResponse.get(count++)); } } @Override protected String getCacheKey() { return "product_info_" + productId; } private static final class BatchCommand extends HystrixCommand<List<ProductInfo>> { public final Collection<CollapsedRequest<ProductInfo, Long>> requests; public BatchCommand(Collection<CollapsedRequest<ProductInfo, Long>> requests) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductInfoService")) .andCommandKey(HystrixCommandKey.Factory.asKey("GetProductInfosCollapserBatchCommand"))); this.requests = requests; } @Override protected List<ProductInfo> run() throws Exception { // 将一个批次内的商品id给拼接在了一起 StringBuilder paramsBuilder = new StringBuilder(""); for(CollapsedRequest<ProductInfo, Long> request : requests) { paramsBuilder.append(request.getArgument()).append(","); } String params = paramsBuilder.toString(); params = params.substring(0, params.length() - 1); // 在这里,我们可以做到什么呢,将多个商品id合并在一个batch内,直接发送一次网络请求,获取到所有的结果 String url = "http://localhost:8082/getProductInfos?productIds=" + params; String response = HttpClientUtils.sendGetRequest(url); List<ProductInfo> 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<Future<ProductInfo>> futures =new ArrayList<Future<ProductInfo>>(); for(String productId:productIds.split(",")){ GetProductInfosCollapser getProductInfosCollapser = new GetProductInfosCollapser(Long.valueOf(productId)); futures.add(getProductInfosCollapser.queue()); } try { for(Future<ProductInfo> 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)));