Hystrix请求合并_hystrixcollapse

应用场景
将短时间内的，仅参数不同的多个请求合并成一个请求，这可以在高并发场景下减少线程数量以及网路连接数量。

好处：

• 在高并发下以及大量重复请求下很有用
• 增加应用的请求吞吐量

缺点

• 会导致延迟大幅提上，具体延迟要看如何配置

批处理Command实现

创建provider提供接口

@RestController
public class UserController {
    @GetMapping("/user/{ids}")
    public List<User> getUserByIds(@PathVariable String ids){
        System.out.println(ids);
        String[] split = ids.split(",");
        List<User> users = new ArrayList<>();
        for (String s : split)
        {
            User user = new User();
            user.setId(Integer.parseInt(s));
            users.add(user);
        }
        return users;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

解释：
创建接口，将1,2,3,4根据","分开，分别赋予四个对象组成集合返回

创建发起合并过后的请求的Service

@Service
public class UserService {

    @Autowired
    RestTemplate restTemplate;

    /**
     * 发送合并起来的请求
     * @param ids
     * @return
     */
    public List<User> getUserByIds(List<Integer> ids){
        User[] users = restTemplate.getForObject("http://provider/user/{1}", User[].class, StringUtils.join(ids, ","));
         return Arrays.asList(users);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

将来将传入的参数集合拼接到url请求中，返回响应结果

创建批处理Command

//批处理，将多个请求合并成一个请求
public class UserBatchCommand  extends HystrixCommand<List<User>> {
    private List<Integer > ids;  //要合并的请求参数
    private UserService userService;  //发送合并请求的Service

    //构造，Setter
    public UserBatchCommand(List<Integer > ids , UserService userService) {
        super(HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("batchcmd")).andCommandKey(HystrixCommandKey.Factory.asKey("batchkey")));
        this.ids = ids;
        this.userService = userService;
    }
    //执行方法
    @Override
    protected List<User> run() throws Exception {
        return userService.getUserByIds(ids);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

解释：创建批处理Command，将从构造方法传进来的ids带入run方法，调用请求合并的Service

最重要的一步就是创建请求合并的HystrixCollapseCommand

public class UserCollapseCommand extends HystrixCollapser<List<User>,User,Integer> {
    private UserService userService;  //在批处理调用合并请求发送Service
    private Integer id;// 每一个请求的参数
    //构造方法
    public UserCollapseCommand( UserService userService, Integer id) {
        super(HystrixCollapser
                .Setter
                .withCollapserKey(HystrixCollapserKey.Factory.asKey("userCollapseCommand"))
                .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(200))); //延时200  默认10ms
        this.userService = userService;
        this.id = id;
    }
    /**
     * 获得请求参数
     * @return
     */
    @Override
    public Integer getRequestArgument() {
        return id;
    }
    /**
     * 请求合并方法
     * @param collection
     * @return
     */
    @Override
    protected HystrixCommand<List<User>> createCommand(Collection<CollapsedRequest<User, Integer>> collection) {
        List<Integer> ids = new ArrayList<>(collection.size()); //创造足够存所有请求参数的List
        for (CollapsedRequest<User, Integer> userIntegerCollapsedRequest : collection) {
            ids.add(userIntegerCollapsedRequest.getArgument());//将参数合并到一起
        }
        return new UserBatchCommand(ids,userService);//传入批处理Command
    }


    /**
     * 请求结果分发
     * @param userList
     * @param collection
     */
    @Override
    protected void mapResponseToRequests(List<User> userList, Collection<CollapsedRequest<User, Integer>> collection) {
        int count = 0;
        for (CollapsedRequest<User, Integer> userIntegerCollapsedRequest : collection) {
            userIntegerCollapsedRequest.setResponse(userList.get(count++));
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

• 将每一个参数通过构造函数传入，重载获取参数的方法getRequestArgument（），直接返回注入的id即可
• 重载合并请求的方法：createCommand，注意到这个方法返回的是HystrixCommand类型，就是刚刚创建的合并请求批处理Command,所以在这个方法里面只要把参数全部拿到，然后合并参数传入合并请求的批处理Command
• 还有一盒就是响应结果分发，也就是给每一个请求一个满意的答复
• 控制请求合并还有另一种方法：maxRequestsInBatch 设置HystrixCollapserProperties.Setter()
• withMaxRequestsInBatch(int value) ，在value中设置请求的最大数量，这种方式只要请求达到预设的数量，就会合并，不然就一等，不推荐这种方式

接下来就是测试方法：

    @GetMapping("/collapseRequest")
    public void collapseRequest() throws ExecutionException, InterruptedException {
        HystrixRequestContext ctx = HystrixRequestContext.initializeContext();
        UserCollapseCommand userCollapseCommand1 =  new UserCollapseCommand(userService, 1);
        UserCollapseCommand userCollapseCommand2 =  new UserCollapseCommand(userService,2);
        UserCollapseCommand userCollapseCommand3 =  new UserCollapseCommand(userService,3);
        Future<User> queue1 = userCollapseCommand1.queue();
        Future<User> queue2 = userCollapseCommand2.queue();
        Future<User> queue3 = userCollapseCommand3.queue();
        User user1 = queue1.get();
        User user2 = queue2.get();
        User user3 = queue3.get();
        Thread.sleep(2000);
        UserCollapseCommand userCollapseCommand4 =  new UserCollapseCommand(userService,96);
        Future<User> queue4 = userCollapseCommand4.queue();
        User user4 = queue4.get();
        System.out.println(user1);
        System.out.println(user2);
        System.out.println(user3);
        System.out.println(user4);
        ctx.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

在中间，调用Thread.sleep(2000)语句暂停了两面，为了就是超过预设的200ms，排除第四个请求到合并请求的队列中；

从结果就很容易看出这样发送了两次请求，参数123 为一组 4为单独一个

注解实现

一个方法两个注解：
在发送合并请求的Service里添加一个方法：

    @HystrixCollapser(batchMethod = "getUserByIds" ,collapserProperties = {@HystrixProperty(name = "timerDelayInMilliseconds",value = "200")})
    public Future<User> getUserById(Integer id){
        return  null;
    }
1
2
3
4

@HystrixCollapser注解里面的属性也是之前Setter里面的相似，这个方法返回Future 返回值 null;

最后还要在发送合并请求的方法上，也就是@HystrixCollapser 的 batchMathod 指出的方法名上添加@HystrixCommand注解即可

测试

于批处理不太一样

直接调用注入的Service调用刚创建的getUserById方法，这个方法直接返回Future 剩下的跟批处理一样;