Three skills of interface request merging, and the performance is directly exploded!

Recommended today

 Reduce  try-catch , This is grace ！

 An addictive new generation of development artifacts , Say goodbye to Controller、Service、Dao Other methods 

SpringBoot Realize face recognition function 

 believe me , Use  Stream  Can really make the code more elegant ！

 The most detailed thread pool in the whole network  ThreadPoolExecutor  Reading ！

 Batch splitting with multithreading  List  Import database , Efficiency bars ！

Merge similar or repeated requests in the upstream system and send them to the downstream system , It can greatly reduce the load of downstream system , Improve the overall system throughput . The article introduces hystrix collapser、ConcurrentHashMultiset、 Self realization BatchCollapser Three request merging techniques , And through its specific implementation, compare their applicable scenarios .

Preface

In the work , Our common request models are ” request - The reply ” type , In one request , The service assigns a separate thread to the request , A separate memory space , All operations are independent , Including resources and system operations . We also know that , Process the system once in the request I/O It's very expensive , If there are very many requests of the same kind I/O operation , So can you put these I/O Operations are merged together , Do it once. I/O operation , Can the burden of downstream resource servers be greatly reduced ？

Recently, I spent most of my spare time exploring this problem , Compared with several existing class libraries , To solve a small problem, put hystrix javanica I turned over the code of , According to the business requirements encountered in my work, I also implemented a simple merge class , The harvest is still great . Maybe this demand is a little ” Pianmen ”, There are not many search results on the Internet , There is no comprehensive information , Simply summarize and share with yourself , I hope I can help my friends who encounter this kind of problem later .

Hystrix Collapser

hystrix

Open source request merge class library （ Well-known ） It seems that only Netflix Company open source Hystrix 了 , hystrix Focus on keeping WEB The system stability of server in high concurrency environment , We often use its fuse (Circuit Breaker) To realize service isolation and disaster time degradation of services , With it , It can prevent the whole system from being washed down by the high concurrency flood of an interface , Even if the interface hangs, the service can be degraded , Return a personalized response . Request consolidation as a sharp weapon to ensure the stability of downstream services , stay hystrix It's no surprise that it's implemented in .

We are using hystrix when , Commonly used javanica modular , Write in the form of annotations hystrix Code , Make the code more concise and less intrusive to business code . Therefore, in the project, we generally need to quote at least hystrix-core and hystrix-javanica Two bags .

in addition ,hystrix All of them are realized through AOP, We have to work on the project xml Explicit configuration in HystrixAspect Of bean To enable it .

<aop:aspectj-autoproxy/>  
<bean id="hystrixAspect" class="com.netflix.hystrix.contrib.javanica.aop.aspectj.HystrixCommandAspect" />

collapser

hystrix collapser yes hystrix Request combiner in , It has custom BatchMethod and Annotation is implemented in two ways , Customize BatchMethod There are various tutorials on the Internet , It's very complicated to implement , A lot of code needs to be written by hand , The annotation method only needs to add two lines of annotations , But I can't find the configuration mode in the official documents , In Chinese, this article should be the only one .

In fact, what we need to pay attention to is ：

• We add... To the methods that need to be merged @HystrixCollapser annotation , Add... To the defined merge method @HystrixCommand annotation ;

• single Method can only pass in one parameter , In the case of multiple parameters, you need to wrap a parameter class by yourself , and batch Method needs  java.util.List<SingleParam>;

• single Method returns  java.util.concurrent.Future<SingleReturn>, batch Method returns  java.util.List<SingleReturn>, And ensure that the number of returned results is consistent with the number of passed in parameters .

Here is a simple example :

public class HystrixCollapserSample {  
  
    @HystrixCollapser(batchMethod = "batch")  
    public Future<Boolean> single(String input) {  
        return null; // single Method will not be executed to   
    }  
  
    public List<Boolean> batch(List<String> inputs) {  
        return inputs.stream().map(it -> Boolean.TRUE).collect(Collectors.toList());  
    }  
}

The source code to achieve

In order to solve hystrix collapser Let's take a look at the configuration of hystrix javanica Source code , Here is a brief summary hystrix Specific implementation of request combiner , The detailed analysis of the source code is in my notes ：Hystrix collasper The source code parsing .

• stay spring-boot To register the facet class bean, It contains @HystrixCollapser Annotation section ;

• The method was detected as being executed HystrixCollapser After the note ,spring call  methodsAnnotatedWithHystrixCommand Method to execute hystrix agent ;

• hystrix Get one collapser example （ At present scope Create if not detected within ）;

• hystrix Submit the parameters of the current request to collapser, from collapser Store in a  concurrentHashMap （RequestArgumentType -> CollapsedRequest） Inside , This method creates a Observable object , And return a Observe the of this object Future To the business thread ;

• collpser When created, a timer Threads , Regular consumption of stored requests ,timer Multiple requests will be constructed into a combined request , call batch After execution, the result sequence is mapped to the output parameters , And notify Future The task has been completed .

We need to pay attention to , Because of the need to wait timer Perform the actual requested operation ,collapser Will cause all requests cost Will increase by about timerInterval/2 ms;

To configure

hystrix collapser The configuration needs to be in @HystrixCollapser Use... On comments , There are two main parts , Proprietary configuration and hystrixCommand General configuration ;

Proprietary configurations include ：

• collapserKey, This can be done without configuration ,hystrix The current method name will be used by default ;

• batchMethod, To configure batch Method name , We usually will single Methods and batch Methods are defined in the same class , Directly fill the name of the method ;

• scope, The most difficult configuration item , It's also the culprit who forced me to read the source code ,com.netflix.hystrix.HystrixCollapser.Scope  Enumeration class , Yes REQUEST, GLOBAL Two options , stay scope by REQUEST when ,hystrix A... Is created for each request collapser, You will find batch Method execution , The total number of incoming requests is 1. and REQUEST Item is still the default item , I don't understand the meaning of such a request for merger ;

• collapserProperties, In this option, we can configure hystrixCommand General configuration of ;

Common configurations include ：

• maxRequestsInBatch, When constructing batch requests , Maximum number of single requests used ;

• timerDelayInMilliseconds, This option configures collapser Of timer How often do threads merge requests ;

• requestCache.enabled, Configure whether to cache when submitting requests ;

A complete configuration is as follows ：

@HystrixCollapser(  
            batchMethod = "batch",  
            collapserKey = "single",  
            scope = com.netflix.hystrix.HystrixCollapser.Scope.GLOBAL,  
            collapserProperties = {  
                    @HystrixProperty(name = "maxRequestsInBatch", value = "100"),  
                    @HystrixProperty(name = "timerDelayInMilliseconds", value = "1000"),  
                    @HystrixProperty(name = "requestCache.enabled", value = "true")  
            })

BatchCollapser

Design

Due to business requirements , We don't care much about the return value of the merged request , And I think hystrix Keep so much Future There's no need to , So I implemented a simple request combiner , Business threads simply put requests into a container , The number of requests has accumulated to a certain amount or delayed for a certain time , Take out the data in the container and send it to the downstream system .

The design idea is similar to hystrix similar , The combiner has a field as a container for storing requests , And set a timer Threads regularly consume requests in the container , The business thread submits the request parameters to the merge In the container of the container . The difference is , After the business thread submits the request to the container, the synchronization returns success , Regardless of the requested consumption results , In this way, the merge trigger in the time dimension is realized .

in addition , I also added another dimension trigger , Every time the request parameter is added to the container, the number of requests in the container will be checked , If the quantity reaches a certain threshold , It will be merged and executed once within the business thread .

Because there are two dimensions that trigger a merge , You will inevitably encounter thread safety problems . In order to ensure that the requests in the container will not be consumed repeatedly or missed by multiple threads , I need a container that can meet the following conditions ：

• It's a kind of Collection, Be similar to ArrayList or Queue, Repeated elements can be stored in order ;

• In a multithreaded environment, it can safely take out all the data inside for consumption , Instead of locking yourself .

java.util.concurrent  The package LinkedBlockingDeque It just meets the requirements , First of all, it implements BlockingDeque Interface , Access operations in a multithreaded environment are safe ; Besides , It also provides  drainTo(Collection<? super E> c, int maxElements) Method , The container can be maxElements Take out the elements safely , Put it in Collection c in .

Realization

The following is the specific code implementation ：

public class BatchCollapser<E> implements InitializingBean {  
     private static final Logger logger = LoggerFactory.getLogger(BatchCollapser.class);  
     private static volatile Map<Class, BatchCollapser> instance = Maps.newConcurrentMap();  
     private static final ScheduledExecutorService SCHEDULE_EXECUTOR = Executors.newScheduledThreadPool(1);  
  
     private volatile LinkedBlockingDeque<E> batchContainer = new LinkedBlockingDeque<>();  
     private Handler<List<E>, Boolean> cleaner;  
     private long interval;  
     private int threshHold;  
  
     private BatchCollapser(Handler<List<E>, Boolean> cleaner, int threshHold, long interval) {  
         this.cleaner = cleaner;  
         this.threshHold = threshHold;  
         this.interval = interval;  
     }  
  
     @Override  
     public void afterPropertiesSet() throws Exception {  
         SCHEDULE_EXECUTOR.scheduleAtFixedRate(() -> {  
             try {  
                 this.clean();  
             } catch (Exception e) {  
                 logger.error("clean container exception", e);  
             }  
         }, 0, interval, TimeUnit.MILLISECONDS);  
     }  
  
     public void submit(E event) {  
         batchContainer.add(event);  
         if (batchContainer.size() >= threshHold) {  
             clean();  
         }  
     }  
  
     private void clean() {  
         List<E> transferList = Lists.newArrayListWithExpectedSize(threshHold);  
         batchContainer.drainTo(transferList, 100);  
         if (CollectionUtils.isEmpty(transferList)) {  
             return;  
         }  
  
         try {  
             cleaner.handle(transferList);  
         } catch (Exception e) {  
             logger.error("batch execute error, transferList:{}", transferList, e);  
         }  
     }  
  
     public static <E> BatchCollapser getInstance(Handler<List<E>, Boolean> cleaner, int threshHold, long interval) {  
         Class jobClass = cleaner.getClass();  
         if (instance.get(jobClass) == null) {  
             synchronized (BatchCollapser.class) {  
                 if (instance.get(jobClass) == null) {  
                     instance.put(jobClass, new BatchCollapser<>(cleaner, threshHold, interval));  
                 }  
             }  
         }  
  
         return instance.get(jobClass);  
     }  
 }

Points to note in the following code ：

• Due to the global requirements of the combiner , You need to implement the combiner as a singleton , In addition, in order to improve its universality , For internal use concurrentHashMap and double check A simple singleton factory is implemented .

• In order to distinguish different purpose combiners , The factory needs to pass in an implementation Handler Example , By way of example class To store requests in groups .

• because  java.util.Timer  The blocking characteristics of , One Timer When another thread starts, it will not block Timer Threads , So use  ScheduledExecutorService  Timing start Timer Threads .

ConcurrentHashMultiset

Design

The request merging described above is to send multiple requests at one time , When the downstream server processes, it is essentially multiple requests , The best request merging is done in memory , Simply combine the request results into one and send it to the downstream server . Such as the demand we often encounter ： Element score accumulation or data statistics , You can first accumulate the score or data of an item in memory , Regularly request the database to save .

Guava Such a data structure is provided in ：ConcurrentHashMultiset, It's different from ordinary set Structure directly overwrites the original element when storing the same element , Instead, keep a count for each element count, When inserting a duplicate element count It's worth adding 1. And it can ensure thread safety without locking when adding and deleting , The concrete implementation is through a  while(true)  Loop try operation , Until the operation is enough .

ConcurrentHashMultiset  This feature of weight removal and counting , It is very suitable for data statistics, which has a high repetition rate in a short time , Calculate the quantity after weight removal , It can greatly reduce the pressure of downstream servers , Even if the repetition rate is not high , Can use a small amount of memory space in exchange for the improvement of system availability , It's also very cost-effective .

Realization

Use  ConcurrentHashMultiset  There is not much difference in the overall structure between request merging and using ordinary containers , The details are similar to ：

if (ConcurrentHashMultiset.isEmpty()) {  
    return;  
}  
  
List<Request> transferList = Lists.newArrayList();  
ConcurrentHashMultiset.elementSet().forEach(request -> {  
    int count = ConcurrentHashMultiset.count(request);  
    if (count <= 0) {  
        return;  
    }  
  
    transferList.add(count == 1 ? request : new Request(request.getIncrement() * count));  
    ConcurrentHashMultiset.remove(request, count);  
});

Summary

Finally, summarize the applicable scenarios of each technology ：

• hystrix collapser: You need the result of each request , And don't care about every request cost Will increase ;

• BatchCollapser: Don't care about the result of the request , Request merging can be triggered in both time and quantity dimensions ;

• ConcurrentHashMultiset： Statistical scenarios with high request repetition rate ;

in addition , If you choose to do it yourself , It's quite possible to  BatchCollapser  and  ConcurrentHashMultiset  Combine it with , stay BatchCollapser Use in  ConcurrentHashMultiset  As a container , In this way, we can combine the advantages of the two

source ：https://zhenbianshu.github.io/

 Last , Recommend a planet of my knowledge , Join now , front  100  name , It only needs  25  Yuan is enough , Very favorable .