[[211861]]

通常微服務架構中的依賴通過遠程調用實現(xiàn)，而遠程調用中最常見的問題就是通信消耗與連接數(shù)占用。在高并發(fā)的情況之下，因通信次數(shù)的增加，總的通信時間消耗將會變的不那么理想。同時，因為對依賴服務的線程池資源有限，將出現(xiàn)排隊等待與響應延遲的情況。為了優(yōu)化這兩個問題，Hystrix提供了HystrixCollapser來實現(xiàn)請求的合并，以減少通信消耗和線程數(shù)的占用。

HystrixCollapser實現(xiàn)了在HystrixCommand之前放置一個合并處理器，它將處于一個很短時間窗（默認10毫秒）內對同一依賴服務的多個請求進行整合并以批量方式發(fā)起請求的功能（服務提供方也需要提供相應的批量實現(xiàn)接口）。通過HystrixCollapser的封裝，開發(fā)者不需要去關注線程合并的細節(jié)過程，只需要關注批量化服務和處理。下面我們從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：請求參數(shù)類型

而對于這三個類型的使用可以在它的三個抽象方法中看到：

• RequestArgumentType getRequestArgument()：該函數(shù)用來定義獲取請求參數(shù)的方法。
• HystrixCommand<BatchReturnType> createCommand(Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests)：合并請求產生批量命令的具體實現(xiàn)方法。
• mapResponseToRequests(BatchReturnType batchResponse, Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests)：批量命令結果返回后的處理，這里需要實現(xiàn)將批量結果拆分并傳遞給合并前的各個原子請求命令的邏輯。

接下來，我們通過一個簡單的示例來直觀的理解實現(xiàn)請求合并的過程。

假設，當前微服務USER-SERVICE提供了兩個獲取User的接口：

• /users/{id}：根據id返回User對象的GET請求接口。
• /users?ids={ids}：根據ids參數(shù)返回User對象列表的GET請求接口，其中ids為以逗號分割的id集合。

而在服務消費端，為這兩個遠程接口已經通過RestTemplate實現(xiàn)了簡單的調用，具體如下：

@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, ",")); 
    } 
 
} 

接著，我們來實現(xiàn)將短時間內多個獲取單一User對象的請求命令進行合并的實現(xiàn)：

• ***步：為請求合并的實現(xiàn)準備一個批量請求命令的實現(xiàn)，具體如下：

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實現(xiàn)，從上面的實現(xiàn)中可以看到它通過調用userService.findAll方法來訪問/users?ids={ids}接口以返回User的列表結果。

• 第二步，通過繼承HystrixCollapser實現(xiàn)請求合并器：

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); 
        } 
    } 
 
} 

在上面的構造函數(shù)中，我們?yōu)檎埱蠛喜⑵髟O置了時間延遲屬性，合并器會在該時間窗內收集獲取單個User的請求并在時間窗結束時進行合并組裝成單個批量請求。下面getRequestArgument方法返回給定的單個請求參數(shù)userId，而createCommand和mapResponseToRequests是請求合并器的兩個核心：

• createCommand：該方法的collapsedRequests參數(shù)中保存了延遲時間窗中收集到的所有獲取單個User的請求。通過獲取這些請求的參數(shù)來組織上面我們準備的批量請求命令
• UserBatchCommand實例。

mapResponseToRequests：在批量命令UserBatchCommand實例被觸發(fā)執(zhí)行完成之后，該方法開始執(zhí)行，其中batchResponse參數(shù)保存了createCommand中組織的批量請求命令的返回結果，而collapsedRequests參數(shù)則代表了每個被合并的請求。在這里我們通過遍歷批量結果batchResponse對象，為collapsedRequests中每個合并前的單個請求設置返回結果，以此完成批量結果到單個請求結果的轉換。

請求合并的原理分析

下圖展示了在未使用HystrixCollapser請求合并器之前的線程使用情況。可以看到當服務消費者同時對USER-SERVICE的/users/{id}接口發(fā)起了五個請求時，會向該依賴服務的獨立線程池中申請五個線程來完成各自的請求操作。

而在使用了HystrixCollapser請求合并器之后，相同情況下的線程占用如下圖所示。由于同一時間發(fā)生的五個請求處于請求合并器的一個時間窗內，這些發(fā)向/users/{id}接口的請求被請求合并器攔截下來，并在合并器中進行組合，然后將這些請求合并成一個請求發(fā)向USER-SERVICE的批量接口/users?ids={ids}，在獲取到批量請求結果之后，通過請求合并器再將批量結果拆分并分配給每個被合并的請求。從圖中我們可以看到以來，通過使用請求合并器有效地減少了對線程池中資源的占用。所以在資源有效并且在短時間內會產生高并發(fā)請求的時候，為避免連接不夠用而引起的延遲可以考慮使用請求合并器的方式來處理和優(yōu)化。

使用注解實現(xiàn)請求合并器

在快速入門的例子中，我們使用@HystrixCommand注解優(yōu)雅地實現(xiàn)了HystrixCommand的定義，那么對于請求合并器是否也可以通過注解來定義呢？答案是肯定！

以上面實現(xiàn)的請求合并器為例，也可以通過如下方式實現(xiàn)：

@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注解為其創(chuàng)建了合并請求器，通過batchMethod屬性指定了批量請求的實現(xiàn)方法為findAll方法（即：請求/users?ids={ids}接口的命令），同時通過collapserProperties屬性為合并請求器設置相關屬性，這里使用@HystrixProperty(name="timerDelayInMilliseconds", value = "100")將合并時間窗設置為100毫秒。這樣通過@HystrixCollapser注解簡單而又優(yōu)雅地實現(xiàn)了在/users/{id}依賴服務之前設置了一個批量請求合并器。

請求合并的額外開銷

雖然通過請求合并可以減少請求的數(shù)量以緩解依賴服務線程池的資源，但是在使用的時候也需要注意它所帶來的額外開銷：用于請求合并的延遲時間窗會使得依賴服務的請求延遲增高。比如：某個請求在不通過請求合并器訪問的平均耗時為5ms，請求合并的延遲時間窗為10ms（默認值），那么當該請求的設置了請求合并器之后，最壞情況下（在延遲時間窗結束時才發(fā)起請求）該請求需要15ms才能完成。

由于請求合并器的延遲時間窗會帶來額外開銷，所以我們是否使用請求合并器需要根據依賴服務調用的實際情況來選擇，主要考慮下面兩個方面：

• 請求命令本身的延遲。如果依賴服務的請求命令本身是一個高延遲的命令，那么可以使用請求合并器，因為延遲時間窗的時間消耗就顯得莫不足道了。
• 延遲時間窗內的并發(fā)量。如果一個時間窗內只有1-2個請求，那么這樣的依賴服務不適合使用請求合并器，這種情況下不但不能提升系統(tǒng)性能，反而會成為系統(tǒng)瓶頸，因為每個請求都需要多消耗一個時間窗才響應。相反，如果一個時間窗內具有很高的并發(fā)量，并且服務提供方也實現(xiàn)了批量處理接口，那么使用請求合并器可以有效的減少網絡連接數(shù)量并極大地提升系統(tǒng)吞吐量，此時延遲時間窗所增加的消耗就可以忽略不計了。

【本文為51CTO專欄作者“翟永超”的原創(chuàng)稿件，轉載請通過51CTO聯(lián)系作者獲取授權】

戳這里，看該作者更多好文