在前文里，我们讲述了通过Hystrix进行容错处理的方式，这里我们将讲述通过Hystrix合并请求的方式

哪怕一个URL请求调用的功能再简单，Web应用服务都至少会开启一个线程来提供服务，换句话说，有效降低URL请求数能很大程度上降低系统的负载。通过Hystrix提供的“合并请求”机制，我们能有效地降低请求数量。

在如下的HystrixMergeDemo.java里，我们将收集2秒内到达的所有“查询订单”的请求，并把它们合并到一个对象中传输给后台，后台则是根据多个请求参数统一返回查询结果，这种基于合并的做法将比每次只处理一个请求的方式要高效得多，代码比较长，我们按类来说明。

1    //省略必要的package和import的代码

2    class OrderDetail{ //订单业务类，其中包含2个属性

3      private String orderId;

4      private String orderOwner;

5      //省略针对orderId和orderOwner的get和set方法

6          //重写toString方法，方便输出

7          public String toString() {

8            return "orderId: " + orderId + ", orderOwner: " + orderOwner ;

9          }

10    }

11    //合并订单请求的处理器

12    class OrderHystrixCollapser extends        HystrixCollapser<Map<String, OrderDetail>, OrderDetail, String>

13    {

14        String orderId;

15        //在构造函数里传入请求参数

16        public OrderHystrixCollapser(String orderId)

17    { this.orderId = orderId;}

18    //指定根据orderId去请求OrderDetail

19        public String getRequestArgument()

20    { return orderId;    }

21    //创建请求命令

22        protected HystrixCommand<Map<String, OrderDetail>> createCommand(

23            Collection<CollapsedRequest<OrderDetail, String>> requests)

24    { return new MergerCommand(requests); }

25        //把请求得到的结果和请求关联到一起

26        protected void mapResponseToRequests(Map<String, OrderDetail> batchResponse,

27            Collection<CollapsedRequest<OrderDetail, String>> requests) {

28            for (CollapsedRequest<OrderDetail, String> request : requests)

29        {

30                // 请注意这里是得到单个请求的结果

31                OrderDetail oneOrderDetail = batchResponse.get(request.getArgument());

32                // 把结果关联到请求中

33                request.setResponse(oneOrderDetail);

34            }

35        }

36    }

在第2行，我们定义了OrderDetail类，这里，我们将合并针对该类对象的请求。

在第12行，我们定义了合并订单的处理器OrderHystrixCollapser类，它继承（extends）了HystrixCollapser<Map<String, OrderDetail>, OrderDetail, String>类，而HystrixCollapser泛型中包含了3个参数，其中第一个参数Map<String, OrderDetail>表示该合并处理器合并请求后返回的结果类型，第二个参数表示是合并OrderDetail类型的对象，第三个参数则表示是根据String类型的请求参数来合并对象。

在第19行里，我们指定了是根据String类型的OrderId参数来请求OrderDetail对象，在第22行的createCommand方法里，我们指定了是调用MergerCommand方法来请求多个OrderDetail，在第26行的mapResponseToRequests方法里，我们是用第28行的for循环，依次把batchResponse对象中包含的多个的查询结果设置到request对象里，由于request是参数requests里的元素，所以执行完第28行的for循环后，requests对象就能关联到合并后的查询结果。

37    class MergerCommand extends HystrixCommand<Map<String, OrderDetail>> {

38        //用orderDB模拟数据库中的数据

39        static HashMap<String,String> orderDB = new HashMap<String,String> ();

40        static     {

41            orderDB.put("1","Peter");

42            orderDB.put("2","Tom");

43            orderDB.put("3","Mike");

44        }

45        Collection<CollapsedRequest<OrderDetail, String>> requests;

46        public MergerCommand(Collection<CollapsedRequest<OrderDetail, String>> requests) {

47            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory

48                    .asKey("mergeDemo")));

49            this.requests = requests;

50        }

51        //在run方法里根据请求参数返回结果

52        protected Map<String, OrderDetail> run() throws Exception {

53            List<String> orderIds = new ArrayList<String>();

54            //通过for循环，整合参数

55            for(CollapsedRequest<OrderDetail, String> request : requests)

56        { orderIds.add(request.getArgument());     }

57            // 调用服务,根据多个订单Id获得多个订单对象

58            Map<String, OrderDetail> ordersHM = getOrdersFromDB(orderIds);

59            return ordersHM;

60        }

61        //用HashMap模拟数据库，从数据库中获得对象

62        private Map<String, OrderDetail> getOrdersFromDB(List<String> orderIds) {

63            Map<String, OrderDetail> result = new HashMap<String, OrderDetail>();

64            for(String orderId : orderIds) {

65                OrderDetail order = new OrderDetail();

66                //这个本该是从数据库里得到，但为了模拟，仅从HashMap里取数据

67                order.setOrderId(orderId);

68                order.setOrderOwner(orderDB.get(orderId) );

69                result.put(orderId, order);

70            }

71            return result;

72        }

73    }

在MergerCommand类的第38到44行里，我们用了orderDB对象来模拟数据库里存储的订单数据。在第46行的构造函数里，我们用传入的requests对象来构建本类里的同名对象，在这个传入的requests对象里，已经包含了合并后的请求。

在第52行的run方法里，我们通过第55行的for循环，依次遍历requests对象，并组装包含请求参数集合的orderIds对象，随后在第58行里，通过getOrdersFromDB方法，根据List类型的orderIds参数，模拟地从数据库里读取数据。

74    public class HystrixMergeDemo{

75        public static void main(String[] args){

76            // 收集2秒内发生的请求，合并为一个命令执行

77            ConfigurationManager.getConfigInstance().setProperty(    "hystrix.collapser.default.timerDelayInMilliseconds", 2000);

78            // 初始化请求上下文

79            HystrixRequestContext context = HystrixRequestContext    .initializeContext();

80            // 创建3个请求合并处理器

81            OrderHystrixCollapser collapser1 = new OrderHystrixCollapser("1");

82            OrderHystrixCollapser collapser2 = new OrderHystrixCollapser("2");

83            OrderHystrixCollapser collapser3 = new OrderHystrixCollapser("3");

84            // 异步执行

85            Future<OrderDetail> future1 = collapser1.queue();

86            Future<OrderDetail> future2 = collapser2.queue();

87            Future<OrderDetail> future3 = collapser3.queue();

88            try {

89                System.out.println(future1.get());

90                System.out.println(future2.get());

91                System.out.println(future3.get());

92            } catch (InterruptedException e) {

93                e.printStackTrace();

94            } catch (ExecutionException e) {

95                e.printStackTrace();

96            }

97            /关闭请求上下文

98            context.shutdown();

99        }

100    }

第74行定义的HystrixMergeDemo类里包含着main方法，在第77行里，我们设置了合并请求的窗口时间是2秒，在第81到83行，创建了3个合并处理器对象，从第85到87行，我们是通过queue 方法，以异步的方式启动了三个处理器，并在第89到91行里，输出了三个处理器返回的结果。这个程序的运行结果如下。

1    orderId: 1, orderOwner: Peter

2    orderId: 2, orderOwner: Tom

3    orderId: 3, orderOwner: Mike

虽然在main方法里，我们发起了3次调用，但由于这些调用是发生在2秒内的，所以会被合并处理，下面我们结合上述针对类和方法的说明，归纳下合并处理3个请求的流程。

步骤一，在代码的81到83行里，我们是通过OrderHystrixCollapser类型的collapser1等三个对象来传入待合并处理的请求，OrderHystrixCollapser类会通过第16行的构造函数，分别接收三个对象传入的orderId参数，并通过第22行的createCommand方法，调用MergerCommand类的方法执行“根据订单Id查订单”的业务。

这里说明下，由于在OrderHystrixCollapser内第16行的getRequestArgument方法里，我们指定了查询参数名是orderId，所以createCommand方法的requests参数，会用orderId来设置查询请求，同时，MergerCommand类中的相关方法也会用该对象来查询OrderDetail信息。

步骤二，由于在createCommand方法里，调用了MergerCommand类的构造函数，所以会触发该类第52行的run方法，在这个方法里，通过第55行和第56行的for循环，把request请求中包含的多个Argument（也就是OrderId）放入到orderIds这个List类型的对象中，随后通过第58行的getOrdersFromDB方法，根据这些orderIds去找对应的OrderDetail对象。

步骤三，在getORdersFromDB方法里，找到对应的多个OrderDetail对象，并组装成Map<String, OrderDetail>类型的result对象返回，然后按调用链的关系，层层返回给OrderHystrixCollapser类。

步骤四，在OrderHystrixCollapser类的mapResponseToRequests方法里，通过for循环，把多次请求的结果组装到requests对象中。由于requests对象是Collection<CollapsedRequest<OrderDetail, String>>类型的，其中用String类型的OrderId关联到了一个OrderDetail对象，所以这里会把合并查询的结果再拆散给3次请求，具体而言，会把3个OrderDetail对象对应地返回给第85行到第87行通过queue调用的3个请求。

这里请注意，虽然通过合并请求的处理方法能降低URL请求的数量，但如果合并后的URL请求数过多，会撑爆掉合并处理器（这里是OrderHystrixCollapser类）的缓存。比如在某项目里，虽然只设置了合并5秒内的请求，但正好赶上秒杀活动，在这个窗口期内的请求数过万，那么就有可能出问题。

所以一般会在上线前，先通过测试确定合并处理器的缓存容量，随后再预估下平均每秒的可能访问数，然后再据此设置合并的窗口时间。

本人之前写的和本文有关的Spring Cloud其它相关文章。

架构师入门：Spring Cloud系列，Hystrix与Eureka的整合

Hystrix针对不可用服务的保护机制以及引入缓存

通过案例了解Hystrix的各种基本使用方式

Ribbon整合Eureka组件，以实现负载均衡

Spring Clould负载均衡重要组件：Ribbon中重要类的用法

架构师入门：搭建双注册中心的高可用Eureka架构（基于项目实战）

架构师入门：搭建基本的Eureka架构（从项目里抽取）

借助Maven入手Spring Boot第一个程序

巴特西

架构师系列文：通过Spring Cloud组件Hystrix合并请求

架构师入门：Spring Cloud系列，Hystrix与Eureka的整合

最新文章

热门文章