Spring Cloud Hystrix的请求合并

通常微服务架构中的依赖通过远程调用实现，而远程调用中最常见的问题就是通信消耗与连接数占用。在高并发的情况之下，因通信次数的增加，总的通信时间消耗将会变的不那么理想。同时，因为对依赖服务的线程池资源有限，将出现排队等待与响应延迟的情况。为了优化这两个问题，Hystrix提供了HystrixCollapser来实现请求的合并，以减少通信消耗和线程数的占用。

HystrixCollapser实现了在HystrixCommand之前放置一个合并处理器，它将处于一个很短时间窗（默认10毫秒）内对同一依赖服务的多个请求进行整合并以批量方式发起请求的功能（服务提供方也需要提供相应的批量实现接口）。通过HystrixCollapser的封装，开发者不需要去关注线程合并的细节过程，只需要关注批量化服务和处理。下面我们从HystrixCollapser的使用实例，对其合并请求的过程一探究竟。

Hystrix的请求合并示例

public abstract class HystrixCollapser<BatchReturnType, ResponseType, RequestArgumentType> implements 
        HystrixExecutable<ResponseType>, HystrixObservable<ResponseType> {
    ...
    public abstract RequestArgumentType getRequestArgument();

    protected abstract HystrixCommand<BatchReturnType> createCommand(Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests);

    protected abstract void mapResponseToRequests(BatchReturnType batchResponse, Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests);
    ...
}

从 HystrixCollapser抽象类的定义中可以看到，它指定了三个不同的类型：

• BatchReturnType：合并后批量请求的返回类型
• ResponseType：单个请求返回的类型
• RequestArgumentType：请求参数类型

而对于这三个类型的使用可以在它的三个抽象方法中看到：

• RequestArgumentTypegetRequestArgument()：该函数用来定义获取请求参数的方法。
• HystrixCommand<BatchReturnType>createCommand(Collection<CollapsedRequest<ResponseType,RequestArgumentType>>requests)：合并请求产生批量命令的具体实现方法。
• mapResponseToRequests(BatchReturnTypebatchResponse,Collection<CollapsedRequest<ResponseType,RequestArgumentType>>requests)：批量命令结果返回后的处理，这里需要实现将批量结果拆分并传递给合并前的各个原子请求命令的逻辑。

接下来，我们通过一个简单的示例来直观的理解实现请求合并的过程。

假设，当前微服务 USER-SERVICE提供了两个获取 User的接口：

• /users/{id}：根据id返回User对象的GET请求接口。
• /users?ids={ids}：根据ids参数返回User对象列表的GET请求接口，其中ids为以逗号分割的id集合。

而在服务消费端，为这两个远程接口已经通过 RestTemplate实现了简单的调用，具体如下：

@Service
public class UserServiceImpl implements UserService {

    @Autowired
    private RestTemplate restTemplate;

    @Override
    public User find(Long id) {
        return restTemplate.getForObject("http://USER-SERVICE/users/{1}", User.class, id);
    }

    @Override
    public List<User> findAll(List<Long> ids) {
        return restTemplate.getForObject("http://USER-SERVICE/users?ids={1}", List.class, StringUtils.join(ids, ","));
    }

}

接着，我们来实现将短时间内多个获取单一User对象的请求命令进行合并的实现：

• 第一步：为请求合并的实现准备一个批量请求命令的实现，具体如下：

public class UserBatchCommand extends HystrixCommand<List<User>> {

    UserService userService;
    List<Long> userIds;

    public UserBatchCommand(UserService userService, List<Long> userIds) {
        super(Setter.withGroupKey(asKey("userServiceCommand")));
        this.userIds = userIds;
        this.userService = userService;
    }

    @Override
    protected List<User> run() throws Exception {
        return userService.findAll(userIds);
    }

}

批量请求命令实际上就是一个简单的HystrixCommand实现，从上面的实现中可以看到它通过调用 userService.findAll方法来访问 /users?ids={ids}接口以返回User的列表结果。

• 第二步，通过继承 HystrixCollapser实现请求合并器：

public class UserCollapseCommand extends HystrixCollapser<List<User>, User, Long> {

    private UserService userService;
    private Long userId;

    public UserCollapseCommand(UserService userService, Long userId) {
        super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("userCollapseCommand")).andCollapserPropertiesDefaults(
                HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(100)));
        this.userService = userService;
        this.userId = userId;
    }

    @Override
    public Long getRequestArgument() {
        return userId;
    }

    @Override
    protected HystrixCommand<List<User>> createCommand(Collection<CollapsedRequest<User, Long>> collapsedRequests) {
        List<Long> userIds = new ArrayList<>(collapsedRequests.size());
        userIds.addAll(collapsedRequests.stream().map(CollapsedRequest::getArgument).collect(Collectors.toList()));
        return new UserBatchCommand(userService, userIds);
    }

    @Override
    protected void mapResponseToRequests(List<User> batchResponse, Collection<CollapsedRequest<User, Long>> collapsedRequests) {
        int count = 0;
        for (CollapsedRequest<User, Long> collapsedRequest : collapsedRequests) {
            User user = batchResponse.get(count++);
            collapsedRequest.setResponse(user);
        }
    }

}

在上面的构造函数中，我们为请求合并器设置了时间延迟属性，合并器会在该时间窗内收集获取单个User的请求并在时间窗结束时进行合并组装成单个批量请求。下面 getRequestArgument方法返回给定的单个请求参数userId，而 createCommand和 mapResponseToRequests是请求合并器的两个核心：

• createCommand：该方法的 collapsedRequests参数中保存了延迟时间窗中收集到的所有获取单个User的请求。通过获取这些请求的参数来组织上面我们准备的批量请求命令 UserBatchCommand实例。
• mapResponseToRequests：在批量命令 UserBatchCommand实例被触发执行完成之后，该方法开始执行，其中 batchResponse参数保存了 createCommand中组织的批量请求命令的返回结果，而 collapsedRequests参数则代表了每个被合并的请求。在这里我们通过遍历批量结果 batchResponse对象，为 collapsedRequests中每个合并前的单个请求设置返回结果，以此完成批量结果到单个请求结果的转换。

请求合并的原理分析

下图展示了在未使用 HystrixCollapser请求合并器之前的线程使用情况。可以看到当服务消费者同时对 USER-SERVICE的 /users/{id}接口发起了五个请求时，会向该依赖服务的独立线程池中申请五个线程来完成各自的请求操作。

而在使用了 HystrixCollapser请求合并器之后，相同情况下的线程占用如下图所示。由于同一时间发生的五个请求处于请求合并器的一个时间窗内，这些发向 /users/{id}接口的请求被请求合并器拦截下来，并在合并器中进行组合，然后将这些请求合并成一个请求发向 USER-SERVICE的批量接口 /users?ids={ids}，在获取到批量请求结果之后，通过请求合并器再将批量结果拆分并分配给每个被合并的请求。从图中我们可以看到以来，通过使用请求合并器有效地减少了对线程池中资源的占用。所以在资源有效并且在短时间内会产生高并发请求的时候，为避免连接不够用而引起的延迟可以考虑使用请求合并器的方式来处理和优化。

使用注解实现请求合并器

在快速入门的例子中，我们使用 @HystrixCommand注解优雅地实现了 HystrixCommand的定义，那么对于请求合并器是否也可以通过注解来定义呢？答案是肯定！

以上面实现的请求合并器为例，也可以通过如下方式实现：

@Service
public class UserService {

    @Autowired
    private RestTemplate restTemplate;

    @HystrixCollapser(batchMethod = "findAll", collapserProperties = {
            @HystrixProperty(name="timerDelayInMilliseconds", value = "100")
    })
    public User find(Long id) {
        return null;
    }

    @HystrixCommand
    public List<User> findAll(List<Long> ids) {
        return restTemplate.getForObject("http://USER-SERVICE/users?ids={1}", List.class, StringUtils.join(ids, ","));
    }
}

@HystrixCommand我们之前已经介绍过了，可以看到这里通过它定义了两个Hystrix命令，一个用于请求 /users/{id}接口，一个用于请求 /users?ids={ids}接口。而在请求 /users/{id}接口的方法上通过 @HystrixCollapser注解为其创建了合并请求器，通过 batchMethod属性指定了批量请求的实现方法为 findAll方法（即：请求 /users?ids={ids}接口的命令），同时通过 collapserProperties属性为合并请求器设置相关属性，这里使用 @HystrixProperty(name="timerDelayInMilliseconds",value="100")将合并时间窗设置为100毫秒。这样通过 @HystrixCollapser注解简单而又优雅地实现了在 /users/{id}依赖服务之前设置了一个批量请求合并器。

请求合并的额外开销

虽然通过请求合并可以减少请求的数量以缓解依赖服务线程池的资源，但是在使用的时候也需要注意它所带来的额外开销：用于请求合并的延迟时间窗会使得依赖服务的请求延迟增高。比如：某个请求在不通过请求合并器访问的平均耗时为5ms，请求合并的延迟时间窗为10ms（默认值），那么当该请求的设置了请求合并器之后，最坏情况下（在延迟时间窗结束时才发起请求）该请求需要15ms才能完成。

由于请求合并器的延迟时间窗会带来额外开销，所以我们是否使用请求合并器需要根据依赖服务调用的实际情况来选择，主要考虑下面两个方面：

• 请求命令本身的延迟。如果依赖服务的请求命令本身是一个高延迟的命令，那么可以使用请求合并器，因为延迟时间窗的时间消耗就显得莫不足道了。
• 延迟时间窗内的并发量。如果一个时间窗内只有1-2个请求，那么这样的依赖服务不适合使用请求合并器，这种情况下不但不能提升系统性能，反而会成为系统瓶颈，因为每个请求都需要多消耗一个时间窗才响应。相反，如果一个时间窗内具有很高的并发量，并且服务提供方也实现了批量处理接口，那么使用请求合并器可以有效的减少网络连接数量并极大地提升系统吞吐量，此时延迟时间窗所增加的消耗就可以忽略不计了。