当前位置：网站首页>Meituan dynamic thread pool practice ideas, open source

Meituan dynamic thread pool practice ideas, open source

2022-07-06 13:51:00 【yanhom】

Hello everyone , Today, let's talk about a more practical topic , Dynamically monitorable thread pool practices , New open source project (DynamicTp) The address is at the end of the article , Welcome to exchange and study .

Write it at the front

A little Java Little friends with programming experience know ,Java The essence of juc package , It's famous Doug Lea Sir Son's masterpiece , Evaluate a programmer Java What's your level , To some extent, it depends on his understanding of juc How do you master some techniques under the package , This is also one of the technical points that must be asked in the interview .

juc The package mainly includes ：

1. Atomic classes （AtomicXXX）
2. Lock class （XXXLock）
3. Thread synchronization class （AQS、CountDownLatch、CyclicBarrier、Semaphore、Exchanger）
4. Task executor class （Executor System class , Including today's protagonist ThreadPoolExecutor）
5. Concurrent collection classes （ConcurrentXXX、CopyOnWriteXXX） Related collection classes
6. Blocking queue class （BlockingQueue Inheritance system class ）
7.Future Related classes
8. Other auxiliary tools

Multithreaded programming scenario , These are all necessary skills , These will help us write high-quality 、 High performance 、 Less bug Code for , At the same time, these are also Java Some difficult technologies in , It needs persistence , Put this to use , Feel the mystery they bring in use .

The above is a simple list of juc Function classification under the package , In this article, we mainly introduce the of dynamically monitored thread pool , Therefore, the specific content will not be discussed , Come and talk alone when you have time . Before reading this article , I hope readers had better have a certain thread pool ThreadPoolExecutor Use experience , Otherwise it will look a little confused .

If you are right about ThreadPoolExecutor Not very familiar with , The following two articles are recommended

javadoop：https://www.javadoop.com/post/java-thread-pool^[1]

Meituan technology blog ：https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html^[2]

background

Use ThreadPoolExecutor Do you have the following pain points in the process ？

1. The code creates a ThreadPoolExecutor, But I don't know how many core parameters are more appropriate
2. Set parameter values based on experience , After going online, it is found that it needs to be adjusted , Change the code and restart the service , Very trouble
3. Thread pool is a black box for developers , Operation cannot be sensed , Until something goes wrong

If you have the above pain points , This article will introduce the dynamic monitorable thread pool （DynamicTp） Maybe it can help you .

If seen ThreadPoolExecutor Source code , You probably know that it actually provides some set Method , You can dynamically modify the corresponding value at run time , These methods are ：

public void setCorePoolSize(int corePoolSize);
public void setMaximumPoolSize(int maximumPoolSize);
public void setKeepAliveTime(long time, TimeUnit unit);
public void setThreadFactory(ThreadFactory threadFactory);
public void setRejectedExecutionHandler(RejectedExecutionHandler handler);

Now most Internet projects actually deploy microservices , Have their own service management system , The distributed configuration center in the microservice component plays the role of dynamically modifying the configuration , Real time roles . So can we dynamically adjust the thread pool parameters at runtime in combination with the configuration center ？ The answer is yes , And the configuration center is relatively highly available , Using it, you don't have to worry too much about problems with configuration push , It can also reduce the difficulty and workload of developing dynamic thread pool components .

Sum up , We summarize the following background

generalized ： stay Java In development , Want to improve system performance , Thread pool is already a 90% The above people will choose to use the basic tools
uncertainty ： Many thread pools may be created in the project , both IO intensive , Also have CPU intensive , But the parameters of thread pool are not easy to determine ; It is necessary to have a set of mechanism to dynamically adjust parameters during operation
Insensibility , The indicators in the running process of thread pool are generally not perceived ; We need a set of monitoring and alarm mechanism in advance 、 In this way, developers can be aware of the running status of the thread pool , Deal with in time
High availability , Configuration changes need to be pushed to the client in time ; Highly available configuration management push service is required , The configuration center is a component that most internet systems now use , Combined with it, it can greatly reduce the amount of development and the difficulty of access

brief introduction

Based on the configuration center, we set the thread pool ThreadPoolExecutor Make some extensions , Realize the dynamic modification of thread pool parameters in operation , In real time ; And monitor the running status of thread pool in real time , Alarm when the set alarm policy is triggered , The alarm information will be pushed to the office platform （ nailing 、 Enterprise, micro, etc ）. Alarm dimensions include （ Queue capacity 、 Thread activity pool 、 Reject trigger, etc ）; At the same time, the thread pool index data will also be collected regularly for the visual use of the monitoring platform . So that we can always sense the load of the thread pool , Adjust in time according to the situation , Avoid problems affecting online business .

    |  __ \                            (_) |__   __|
    | |  | |_   _ _ __   __ _ _ __ ___  _  ___| |_ __  
    | |  | | | | | '_ \ / _` | '_ ` _ | |/ __| | '_ \ 
    | |__| | |_| | | | | (_| | | | | | | | (__| | |_) |
    |_____/ __, |_| |_|__,_|_| |_| |_|_|___|_| .__/ 
             __/ |                              | |    
            |___/                               |_|    
     :: Dynamic Thread Pool ::

characteristic

Refer to meituan thread pool practice , Thread pool dynamic parameterization management , Increase monitoring 、 Alarm function
be based on Spring frame , Now only support SpringBoot Project use , Lightweight , introduce starter Ready to eat
Dynamic adjustment of thread pool parameters based on configuration center , In real time ; Integrate mainstream configuration center , The default support Nacos、Apollo, It also provides SPI The interface can be customized and extended
Built in notification and alarm function , Provide multiple alarm dimensions （ Configuration change notification 、 Active alarm 、 Capacity threshold alarm 、 The reject policy triggers an alarm ）, Enterprise wechat is supported by default 、 Nail alarm , At the same time provide SPI The interface can be customized and extended
Built in thread pool indicator collection function , Supported by MicroMeter、JsonLog Log output 、Endpoint Three ways , It can be done by SPI Interface custom extension implementation
Integrated management of thread pools of common third-party components , Integrated SpringBoot built-in WebServer（Tomcat、Undertow、Jetty） Thread pool management

Architecture design

It is mainly divided into four modules

Configuration change monitoring module ：
1. Listen to the specified profile of a specific configuration center （ Default implementation Nacos、Apollo）, Available internally SPI Interface extension other implementations
2. Parse the contents of the configuration file , Built in implementation yml、properties Profile resolution , Available internally SPI Interface extension other implementations
3. Notify the thread pool management module to refresh
Thread pool management module ：
1. Pull configuration information from the configuration center when the service is started , The generated thread pool instance is registered in the internal thread pool registry
2. When the monitoring module monitors the configuration change , Pass the change information to the management module , Refresh thread pool parameters
3. Passed in code getExecutor() Method to obtain the thread pool object instance according to the thread pool name
Monitoring module ：
Realize the collection and output of monitoring indicators , The following three methods are provided by default , It can also be provided internally SPI Interface extension other implementations
1. Default implementation Json log Output to disk
2.MicroMeter collection , introduce MicroMeter Related dependencies
3. Thunderstorm Endpoint Endpoint , It can be done by http Access to
Notify the alarm module ：
Connect with the office platform , Realize the notification and alarm function , The default implementation is nailing 、 Enterprise micro , Available internally SPI Interface extension other implementations , The types of notification alarms are as follows
1. Thread pool parameter change notification
2. Blocking queue capacity reaches the set threshold alarm
3. Alarm when thread pool activity reaches the set threshold
4. Trigger the reject policy alarm

Use

maven rely on

apollo Application access depends on this

    <dependency>
        <groupId>io.github.lyh200</groupId>
        <artifactId>dynamic-tp-spring-boot-starter-apollo</artifactId>
        <version>1.0.0</version>
    </dependency>

spring-cloud In the scene nacos Application access depends on this

    <dependency>
        <groupId>io.github.lyh200</groupId>
        <artifactId>dynamic-tp-spring-cloud-starter-nacos</artifactId>
        <version>1.0.0</version>
    </dependency>

Not spring-cloud In the scene nacos Application access depends on this

    <dependency>
        <groupId>io.github.lyh200</groupId>
        <artifactId>dynamic-tp-spring-boot-starter-nacos</artifactId>
        <version>1.0.0</version>
    </dependency>

Thread pool configuration

spring:
  dynamic:
    tp:
      enabled: true
      enabledBanner: true        #  Open or not banner Print , Default true
      enabledCollect: false      #  Whether to start monitoring indicator collection , Default false
      collectorType: logging     #  Monitoring data collector type （JsonLog | MicroMeter）, Default logging
      logPath: /home/logs        #  Monitoring log data path , Default ${user.home}/logs
      monitorInterval: 5         #  Monitoring intervals （ Alarm judgment 、 Index collection ）, Default 5s
      nacos:                     # nacos To configure , No default values are configured （ The rules name-dev.yml such ）
        dataId: dynamic-tp-demo-dev.yml
        group: DEFAULT_GROUP
      apollo:                    # apollo To configure , Default settings are not configured apollo Configure the first namespace
        namespace: dynamic-tp-demo-dev.yml
      configType: yml            #  Profile type 
      platforms:                 #  Notify alarm platform configuration 
        - platform: wechat
          urlKey: 3a7500-1287-4bd-a798-c5c3d8b69c  #  Replace 
          receivers: test1,test2                   #  The name of the recipient's wechat 
        - platform: ding
          urlKey: f80dad441fcd655438f4a08dcd6a     #  Replace 
          secret: SECb5441fa6f375d5b9d21           #  Replace , Not sign The mode may not have this value 
          receivers: 15810119805                   #  Nail account, mobile phone number     
      tomcatTp:                                    # tomcat web server Thread pool configuration 
          minSpare: 100
          max: 400      
      jettyTp:                                     # jetty web server Thread pool configuration 
          min: 100
          max: 400     
      undertowTp:                                  # undertow web server Thread pool configuration 
          ioThreads: 100
          workerThreads: 400      
      executors:                                   #  Dynamic thread pool configuration 
        - threadPoolName: dynamic-tp-test-1
          corePoolSize: 6
          maximumPoolSize: 8
          queueCapacity: 200
          queueType: VariableLinkedBlockingQueue   #  Task queue , View source code QueueTypeEnum Enumeration class 
          rejectedHandlerType: CallerRunsPolicy    #  Refusal strategy , see RejectedTypeEnum Enumeration class 
          keepAliveTime: 50
          allowCoreThreadTimeOut: false
          threadNamePrefix: test           #  Thread name prefix 
          notifyItems:                     #  Alarm item , If not configured, it will be automatically configured （ Change notice 、 Capacity alarm 、 Active alarm 、 Reject alarm ）
            - type: capacity               #  Alarm item type , View source code  NotifyTypeEnum Enumeration class 
              enabled: true
              threshold: 80                #  Alarm threshold 
              platforms: [ding,wechat]     #  Optional configuration , Do not configure the upper layer by default platforms All configured platforms 
              interval: 120                #  Alarm interval （ Company ：s）
            - type: change
              enabled: true
            - type: liveness
              enabled: true
              threshold: 80
            - type: reject
              enabled: true
              threshold: 1

Code generation , Service startup will automatically register

@Configuration
public class DtpConfig {

   @Bean
   public DtpExecutor demo1Executor() {
       return DtpCreator.createDynamicFast("demo1-executor");
  }

   @Bean
   public ThreadPoolExecutor demo2Executor() {
       return ThreadPoolBuilder.newBuilder()
              .threadPoolName("demo2-executor")
              .corePoolSize(8)
              .maximumPoolSize(16)
              .keepAliveTime(50)
              .allowCoreThreadTimeOut(true)
              .workQueue(QueueTypeEnum.SYNCHRONOUS_QUEUE.getName(), null, false)
              .rejectedExecutionHandler(RejectedTypeEnum.CALLER_RUNS_POLICY.getName())
              .buildDynamic();
  }
}

Code calls , Get according to the name of the route pool

public static void main(String[] args) {
       DtpExecutor dtpExecutor = DtpRegistry.getExecutor("dynamic-tp-test-1");
       dtpExecutor.execute(() -> System.out.println("test"));
}

matters needing attention

The parameters configured by the configuration file will override the parameters configured by code generation
The blocking queue has only VariableLinkedBlockingQueue The type can be modified capacity, This type of function and LinkedBlockingQueue be similar , It's just capacity No final type , You can modify , VariableLinkedBlockingQueue Reference resources RabbitMq The implementation of the

After startup, see the following log output to prove that the access is successful


|  __ \                            (_) |__   __|   
| |  | |_   _ _ __   __ _ _ __ ___  _  ___| |_ __  
| |  | | | | | '_ \ / _` | '_ ` _ | |/ __| | '_ \ 
| |__| | |_| | | | | (_| | | | | | | | (__| | |_) |
|_____/ __, |_| |_|__,_|_| |_| |_|_|___|_| .__/ 
         __/ |                              | |    
        |___/                               |_|    
 :: Dynamic Thread Pool :: 

DynamicTp register, executor: DtpMainPropWrapper(dtpName=dynamic-tp-test-1, corePoolSize=6, maxPoolSize=8, keepAliveTime=50, queueType=VariableLinkedBlockingQueue, queueCapacity=200, rejectType=RejectedCountableCallerRunsPolicy, allowCoreThreadTimeOut=false)

Configuration changes push notification messages , And the changed field will be highlighted


DynamicTp [dynamic-tp-test-2] refresh end, changed keys: [corePoolSize, queueCapacity], corePoolSize: [6 => 4], maxPoolSize: [8 => 8], queueType: [VariableLinkedBlockingQueue => VariableLinkedBlockingQueue], queueCapacity: [200 => 2000], keepAliveTime: [50s => 50s], rejectedType: [CallerRunsPolicy => CallerRunsPolicy], allowsCoreThreadTimeOut: [false => false]

Notify the police

Triggering the alarm threshold will push the corresponding alarm message , The relevant fields are highlighted , Active alarm 、 Capacity alert 、 Reject alarm

Configuration changes push notification messages , And the changed field will be highlighted

Monitoring log

Through the main configuration file collectType Attribute to configure the indicator collection type , The default value is ：logging

micrometer The way ： By introducing micrometer Relevant dependencies are collected to the corresponding platform （ Such as Prometheus,InfluxDb...）

logging： The index data is expressed in json Output log to disk in format , Address {appName}.monitor.log

2022-01-16 15:25:20.599 INFO [dtp-monitor-thread-1:d.m.log] {"activeCount":2,"queueSize":100,"largestPoolSize":4,"poolSize":2,"rejectHandlerName":"CallerRunsPolicy","queueCapacity":1024,"fair":false,"rejectCount":0,"waitTaskCount":10,"taskCount":120,"queueRemainingCapacity":1024,"corePoolSize":6,"queueType":"VariableLinkedBlockingQueue","completedTaskCount":1078,"dtpName":"remoting-call","maximumPoolSize":8}
2022-01-16 15:25:25.603 INFO [dtp-monitor-thread-1:d.m.log] {"activeCount":2,"queueSize":120,"largestPoolSize":4,"poolSize":2,"rejectHandlerName":"CallerRunsPolicy","queueCapacity":1024,"fair":false,"rejectCount":0,"waitTaskCount":20,"taskCount":140,"queueRemainingCapacity":1024,"corePoolSize":6,"queueType":"VariableLinkedBlockingQueue","completedTaskCount":1459,"dtpName":"remoting-call","maximumPoolSize":8}
2022-01-16 15:25:30.609 INFO [dtp-monitor-thread-1:d.m.log] {"activeCount":2,"queueSize":140,"largestPoolSize":4,"poolSize":2,"rejectHandlerName":"CallerRunsPolicy","queueCapacity":1024,"fair":false,"rejectCount":0,"waitTaskCount":89,"taskCount":180,"queueRemainingCapacity":1024,"corePoolSize":6,"queueType":"VariableLinkedBlockingQueue","completedTaskCount":1890,"dtpName":"remoting-call","maximumPoolSize":8}
2022-01-16 15:25:35.613 INFO [dtp-monitor-thread-1:d.m.log] {"activeCount":2,"queueSize":160,"largestPoolSize":4,"poolSize":2,"rejectHandlerName":"CallerRunsPolicy","queueCapacity":1024,"fair":false,"rejectCount":0,"waitTaskCount":99,"taskCount":230,"queueRemainingCapacity":1024,"corePoolSize":6,"queueType":"VariableLinkedBlockingQueue","completedTaskCount":2780,"dtpName":"remoting-call","maximumPoolSize":8}
2022-01-16 15:25:40.616 INFO [dtp-monitor-thread-1:d.m.log] {"activeCount":2,"queueSize":230,"largestPoolSize":4,"poolSize":2,"rejectHandlerName":"CallerRunsPolicy","queueCapacity":1024,"fair":false,"rejectCount":0,"waitTaskCount":0,"taskCount":300,"queueRemainingCapacity":1024,"corePoolSize":6,"queueType":"VariableLinkedBlockingQueue","completedTaskCount":4030,"dtpName":"remoting-call","maximumPoolSize":8}

expose EndPoint Endpoint (dynamic-tp), Can pass http Mode request

[
    {
        "dtp_name": "remoting-call",
        "core_pool_size": 8,
        "maximum_pool_size": 16,
        "queue_type": "SynchronousQueue",
        "queue_capacity": 0,
        "queue_size": 0,
        "fair": false,
        "queue_remaining_capacity": 0,
        "active_count": 2,
        "task_count": 2760,
        "completed_task_count": 2760,
        "largest_pool_size": 16,
        "pool_size": 8,
        "wait_task_count": 0,
        "reject_count": 12462,
        "reject_handler_name": "CallerRunsPolicy"
    },
    {
        "max_memory": "220 MB",
        "total_memory": "140 MB",
        "free_memory": "44 MB",
        "usable_memory": "125 MB"
    }
]