当前位置：网站首页>Discussion on related configuration of thread pool

Discussion on related configuration of thread pool

2022-07-02 02:51:00 【Bug Commander】

Reference material

Thread pool

Thread pool is a kind of “ Pooling ” Thread usage mode , By creating a certain number of threads , Keep these threads in a ready state to improve system response speed , After the thread is used, it is returned to the thread pool to achieve the goal of reuse , So as to reduce the consumption of system resources .

Pool benefits

Use thread pool , It has the following advantages

Reduce resource consumption
- Reduces the cost of thread creation and destruction by reusing created threads
Improve response time
- When the mission arrives , Tasks can be executed immediately without waiting for a thread to be created
Improve the manageability of threads
- Threads are scarce resources , If unlimited creation , Not only does it consume system resources , It also reduces the stability of the system , Uniform allocation is possible using thread pools , Tune and monitor
More and more powerful functions
- Thread pool has scalability , Allow developers to add more functionality to it . For example, delay timed thread pool ScheduledThreadPoolExecutor, To allow tasks to be postponed or performed on a regular basis

Executors

In order to better control multithreading ,JDK A set of Executor frame , Its essence is a thread pool , Its core members are as follows .

Interface or class	explain
`Executor` Interface	Defines a receive `Runnable` Object method `executor`
`ExecutorService` Interface	A ratio `Executor` Use a broader subclass interface , It provides a life cycle management approach , And can track the execution status of one or more asynchronous tasks `Future` Methods
`AbstractExecutorService` abstract class	`ExecutorService` The default implementation of the execution method
`ScheduledExecutorService` Interface	An interface that can schedule tasks regularly
`ScheduledThreadPoolExecutor` class	`ScheduledExecutorService` The implementation of the , A thread pool that can schedule tasks regularly
`ThreadPoolExecutor` class	It is mainly used to create thread pools

Common methods

Executors Common methods are as follows

newCachedThreadPool()
- Create a cacheable thread pool
- CachedThreadPool It is suitable for concurrent execution of a large number of short-term and time-consuming tasks , Or a lighter load server
newFiexedThreadPool(int nThreads)
- Creates a thread pool with a fixed number of threads
- FiexedThreadPool It is suitable for scenarios with slightly heavy load but not too many tasks , Scenarios where the number of threads needs to be limited in order to make rational use of resources
newSingleThreadExecutor()
- Create a single threaded Executor
- SingleThreadExecutor It is suitable for serial execution of tasks , Each task is performed in sequence , No concurrent execution required
newScheduledThreadPool(int corePoolSize)
- Create a thread pool that supports scheduled and periodic task execution , In most cases it can be used instead Timer class
- ScheduledThreadPool It's a scheduling pool , Its implementation schedule、scheduleAtFixedRate、scheduleWithFixedDelay Three methods , Delayed execution can be achieved 、 Cycle execution and other operations
newSingleThreadScheduledExecutor()
- Create a corePoolSize by 1 Of ScheduledThreadPoolExecutor
newWorkStealingPool(int parallelism)
- Return to one ForkJoinPool example
- ForkJoinPool Mainly used to realize “ Divide and rule ” The algorithm of , Suitable for computing intensive tasks

Avoid using Executors Creating a thread pool

According to Ali 《Java Development Manual 》, Avoid using Executors Creating a thread pool , Recommended ThreadPoolExecutors Creating a thread pool .

Executors Created FiexedThreadPool and SingleThreadPool The task queue length is Integer.MAX_VALUE, A large number of requests may pile up , Which leads to OOM;
Executors Created CachedThreadPool and ScheduledThreadPool The number of threads allowed to be created is Integer.MAX_VALUE, A large number of threads may be created , Which leads to OOM.

ThreadPoolExecutor

Java in , The implementation class of thread pool is ThreadPoolExecutor, The constructor is as follows .

public ThreadPoolExecutor(int corePoolSize,
                        int maximumPoolSize,
                        long keepAliveTime,
                        TimeUnit timeUnit,
                        BlockingQueue<Runnable> workQueue,
                        ThreadFactory threadFactory,
                        RejectedExecutionHandler handler)

corePoolSize： Number of core threads in thread pool
maximumPoolSize： The maximum number of threads that the thread pool can hold
keepAliveTime： Thread idle lifetime
timeUnit： A unit of time for a thread to survive idle , Such as TimeUnit.MILLISECONDS、TimeUnit.SECONDS
blockingQueue： Task queue , Common task queues include
- ArrayBlockingQueue： A bounded blocking queue implemented by an array , This queue is based on FIFO The principle of sorting elements , Supports fair access to queues
- LinkedBlockingQueue： An optional bounded blocking queue composed of linked list structure , If you don't specify the size , Then use Integer.MAX_VALUE As the queue size , according to FIFO The principle of sorting elements
- PriorityBlockingQueue ： An unbounded blocking queue supporting priority sorting , By default, it is arranged in natural order , You can also specify Comparator
- DelayQueue： An unbounded blocking queue that supports delayed retrieval of elements , When creating an element, you can specify how long it will take to get the current element from the queue , It is often used in cache system design and scheduled task scheduling
- SynchronousQueue： A blocked queue that does not store elements , The save operation must wait for the get operation , vice versa
- LinkedTransferQueue： An unbounded blocking queue composed of linked list structure , And LinkedBlockingQueue Much more transfer and tryTranfer Method , This method will immediately pass the element to the consumer when there is a consumer waiting to receive the element
- LinkedBlockingDeque： A double ended blocking queue composed of linked list structure , You can insert and delete elements from both ends of the queue
threadFactory： Thread factory , Use to specify how to create new threads for the thread pool ,threadFactory You can set the thread name 、 Thread group 、 Priority and other parameters , Such as through Google The toolkit can set the thread name in the thread pool , The code is as follows .

new ThreadFactoryBuilder().setNameFormat("general-detail-batch-%d").build();

RejectedExecutionHandler： Refusal strategy , Common rejection strategies include
- ThreadPoolExecutor.AbortPolicy： The default policy , Thrown when the task queue is full RejectedExecutionException abnormal
- ThreadPoolExecutor.DiscardPolicy： New tasks cannot be discarded , Don't throw any exceptions
- ThreadPoolExecutor.CallerRunsPolicy： When the task queue is full, use the caller's thread to execute the task directly
- ThreadPoolExecutor.DiscardOldestPolicy： When the task queue is full, discard the task blocking the head of the queue （ The oldest task ）, Then add the current task

State of thread

//Thread.State  Source code 
public enum State {
    NEW,
    RUNNABLE,
    BLOCKED,
    WAITING,
    TIMED_WAITING,
    TERMINATED;
}

Java Thread 6 The transition of States is shown in the following figure .

The state of the thread pool

public class ThreadPoolExecutor extends AbstractExecutorService{
   private static final int COUNT_BITS = Integer.SIZE - 3;
   private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

   // runState is stored in the high-order bits
   private static final int RUNNING    = -1 << COUNT_BITS;
   private static final int SHUTDOWN   =  0 << COUNT_BITS;
   private static final int STOP       =  1 << COUNT_BITS;
   private static final int TIDYING    =  2 << COUNT_BITS;
   private static final int TERMINATED =  3 << COUNT_BITS;
}

ThreadPoolExecutor Thread pools have the following states

RUNNING： Running state , Take on new tasks , Continuously process the tasks in the task queue ;
SHUTDOWN： Stop taking on new tasks , But you have to deal with the tasks in the task queue ;
STOP： Stop taking on new tasks , No longer process tasks in the task queue , Interrupt an ongoing task ;
TIDYING： Indicates that the thread pool is stopping , Abort all tasks , Destroy all worker threads , When the thread pool executes terminated() Method when entering TIDYING state ;
TERMINATED： Indicates that the thread pool has stopped functioning , All worker threads have been destroyed , All tasks have been cleared or executed ,terminated() Method execution complete ;

Status of thread pool running , It's not explicitly set by the user , But with the running of thread pool , To maintain by the interior . A variable is used inside the thread pool to maintain two values ： Running state （runState） And the number of threads （workerCount）. In the implementation , Thread pool will run （runState）、 Number of threads （workerCount） The maintenance of two key parameters is put together , As shown in the following code .

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl It's a AtomicInteger Variable of type , It is a field to control the running state of thread pool and the number of effective threads in thread pool , It contains two parts of information , The running state of the thread pool （runState） And the number of valid threads in the thread pool （workerCount）.ctl The height of 3 Bit save runState, low 29 Bit save workerCount. There is no interference between the two variables . Use a variable to store two values , Avoid making relevant decisions , There are inconsistencies , There's no need to maintain consistency , And take up lock resources .

About getting lifecycle state of internal encapsulation 、 Get the calculation method of the number of threads in the thread pool , As shown in the following code .

// Calculate the current running state 
private static int runStateOf(int c){ 
   return c & ~CAPACITY;
} 
// Count the current number of threads 
private static int workerCountOf(int c) { 
   return c & CAPACITY; 
}  
// Generated by state and number of threads ctl
private static int ctlOf(int rs, int wc){ 
   return rs | wc; 
}

Thread pool task scheduling mechanism

When the thread pool submits a task , The main steps of task scheduling are as follows

When the number of surviving core threads in the thread pool is less than corePoolSize The value of the number of core threads parameter , Thread pool will create a core thread to handle the submitted tasks ;
If the number of core threads in the thread pool is full , That is, the number of threads is equal to corePoolSize, The newly submitted task will be tried to put into the task queue workQueue Waiting for execution ;
When the number of threads in the thread pool is equal to corePoolSize 了 , And the task queue workQueue Is full , Then judge whether the current number of threads has reached maximumPoolSize, That is, whether the maximum number of threads is full ; If it doesn't arrive , Create a non core thread to perform the submitted task ;
If the current number of threads has reached maximumPoolSize, When new tasks are submitted , Execute the rejection policy for processing .

As shown in the figure above , Thread pool actually builds a producer consumer model inside , Decouple thread and task , Not directly related , So a good buffer task , Reuse threads . The running of thread pool is mainly divided into two parts

task management
Thread management

「 task management 」 Partly as a producer , When the task is submitted , The thread pool will judge the subsequent flow of the task

Directly request the thread to perform the task
Buffer to queue for thread execution
Reject the task

「 Thread management 」 Some are consumers , They are uniformly maintained in the process pool , According to the task request for the allocation of threads , When the thread finishes executing tasks, it will continue to acquire new tasks to execute , Finally, when the thread cannot get the task , Threads will be recycled .

How many threads are appropriate to create

ref 1-《Java Concurrent programming practice 》
ref 2- Sun Gong is a soldier -Java Concurrent programming Atlas 2022

about CPU Intensive program , The best number of threads = CPU Check the number + 1.

Computationally intensive （CPU intensive ） The thread of ⽣⼀ individual ⻚ Wrong or for other reasons ⽽ Pause , There's just ⼀ individual “ additional ” The thread of , Can ensure that in this case CPU The cycle will not be interrupted ⼯ do .—— 《Java Concurrent programming practice 》

about I/O Intensive program , The best number of threads = CPU The core number * (1 / CPU utilization ) = CPU The core number * (1 + (IO Time consuming / CPU Time consuming ))

Refer to the above formula , For example, almost all the programs are IO Time consuming , Then the optimal number of threads = 2N（N yes CPU Check the number ） or 2N + 1.

Thread pool monitoring

ref 1- How to monitor thread pools

about 「 Thread pool monitoring 」, It mainly involves the following aspects

How to monitor operation data
Monitoring indicators
Monitoring data storage
How to abstract thread pool monitoring into a public service

How to monitor operation data

How to monitor operation data , There are two ways of thinking

Thread pool runtime buried point , Statistics are made for each running task
Get the running data of the thread pool regularly

Recommended use of 2 Ways of planting , Because in the first way , The state of the thread pool will be obtained by obtaining the lock , Relatively poor performance .

Getting the status of the thread pool （ Such as the current number of threads 、 Number of active threads 、 Maximum number of threads present 、 The total number of thread pool tasks completed ） when , We'll get... First mainLock, Then we start to calculate .mainLock Is the master lock of the thread pool , Threads execute 、 This lock is used for thread destruction and thread pool stop .

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
   // Get the status of the thread pool 
   ...
} finally {
    mainLock.unlock();
}

If you get this lock frequently , This will affect the performance of the original thread pool tasks .

Monitoring indicators

indicators	describe
Thread pool current load	Current number of threads / Maximum number of threads
Thread pool peak load	Maximum load during thread pool operation
Number of core threads	Number of core threads in thread pool
Maximum number of threads	The thread pool limits the number of threads that exist at the same time
Current number of threads	Number of threads in the current thread pool
Number of active threads	The approximate number of threads executing the task
Maximum number of threads present	The maximum number of concurrent threads in the thread pool since it was run
Blocking queues	The container for the thread pool to stage tasks
Queue capacity	The maximum number of elements allowed in the queue
Queue element	The number of elements stored in the queue
The remaining capacity of the queue	The number of elements that can still be stored in the queue
Total number of thread pool tasks completed	Approximate total number of tasks completed
Reject policy execution times	The total number of rejections thrown by the runtime

For a thread pool , Callable threadPoolExecutor Relevant methods provided , Get the current number of queued threads of the thread pool 、 Number of currently active threads 、 The number of execution completion threads 、 Bus number

// Current number of queued threads 
int queueSize = threadPoolExecutor.getQueue().size();

// Number of currently active threads 
int activeCount = threadPoolExecutor.getActiveCount();

// The number of execution completion threads 
long completeTaskCount = threadPoolExecutor.getCompletedTaskCount();

// Bus number 
long taskCount = threadPoolExecutor.getTaskCount();

Monitoring data storage

In order to locate the problem , Help developers troubleshoot problems , You need to consider storing the monitoring data , Such as ES To store or MySQL Storage . The information to be stored is as follows （ With MySQL For example ）.

CREATE TABLE `his_run_data` (
  `thread_pool_id` varchar(56) DEFAULT NULL COMMENT ' Thread pool ID',
  `instance_id` varchar(256) DEFAULT NULL COMMENT ' example ID',
  `current_load` bigint(20) DEFAULT NULL COMMENT ' The current load ',
  `peak_load` bigint(20) DEFAULT NULL COMMENT ' Peak load ',
  `pool_size` bigint(20) DEFAULT NULL COMMENT ' Number of threads ',
  `active_size` bigint(20) DEFAULT NULL COMMENT ' Number of active threads ',
  `queue_capacity` bigint(20) DEFAULT NULL COMMENT ' Queue capacity ',
  `queue_size` bigint(20) DEFAULT NULL COMMENT ' Queue element ',
  `queue_remaining_capacity` bigint(20) DEFAULT NULL COMMENT ' The remaining capacity of the queue ',
  `completed_task_count` bigint(20) DEFAULT NULL COMMENT ' Completed task count ',
  `reject_count` bigint(20) DEFAULT NULL COMMENT ' Rejection times ',
  `timestamp` bigint(20) DEFAULT NULL COMMENT ' Time stamp ',
  `gmt_create` datetime DEFAULT NULL COMMENT ' Creation time ',
  `gmt_modified` datetime DEFAULT NULL COMMENT ' Modification time ',
  PRIMARY KEY (`id`),
  KEY `idx_group_key` (`tp_id`,`instance_id`) USING BTREE,
  KEY `idx_timestamp` (`timestamp`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT=' Historical operation data table ';

Abstract as a public service

On the basis of the above analysis , Consider abstracting the monitoring thread pool as a public service . stay 「 Thread pool monitoring service 」 in , Complete the following steps

The client collects thread pool historical running data regularly , Package the data and send it to the server ;
The server receives the data reported by the client , Data warehousing and persistent storage ;
The server regularly deletes or archives the historical running data of the client thread pool ;
The server provides the data display of thread pool running chart .

ad locum , It is recommended that the 「 collection 、 Report 」 and 「 Storage 、 file 」 Separation , The collected monitoring data can be MQ Send to 「 Storage 、 file 」 Server side . Good design , A single responsibility should be ensured .

Dynamically observable thread pool

stay 「 Thread pool monitoring 」 On the basis of , Introduce the dynamic observable thread pool .

The core idea of dynamically observable thread pool is , Configure through the center （ Such as Nacos、Zookeeper） Set the core parameters of the thread pool corePoolSize、maximumPoolSize、workQueue Dynamic match , Data that can be monitored through traffic or thread pools , Feedback adjustment of thread pool .

ThreadPoolExecutor The following methods are provided to adjust the parameters of the thread pool .


public void setCorePoolSize(int corePoolSize);

public void setKeepAliveTime(long time, TimeUnit unit);

public void setMaximumPoolSize(int maximumPoolSize);

public void setRejectedExecutionHandler(RejectedExecutionHandler handler);

public void setThreadFactory(ThreadFactory threadFactory);

The industry is right 「 Dynamically observable thread pool 」 There are also mature solutions , such as

Meituan dynamic thread pool design
DynamicTp： Lightweight dynamic thread pool based on configuration center
Hippo4J： Dynamic observable thread pool framework

Hippo4J Is a dynamic observable thread pool framework , It can improve the online operation guarantee capability for the business system .Hippo4J be based on 「 Meituan dynamic thread pool 」 Design concept development , Dynamic parameter invocation is enhanced for thread pool 、 monitor 、 Alarm function .

The main function

Hippo4J Its main functions include

Dynamic change
- Application runtime dynamically changes thread pool parameters , Including but not limited to the number of cores 、 Maximum number of threads 、 Block queue size and reject policy
- Support differentiated configuration of the thread pool in the cluster
Custom alarms
- Thread pool is embedded when running , Collect monitoring information
- Provide 4 Alarm dimensions , Thread pool overload 、 Blocking queue capacity 、 Run over length and reject policy alarm
Operation monitoring
- Support thread pool running data collection and storage with customized duration
- Provide visual large screen monitoring operation indicators

modular	hippo4j-core	hippo4j-server
rely on	Nacos、Apollo Wait for the configuration center	Deploy Hippo4J Server（ No internal dependent middleware ）
Use	The configuration center supplements thread pool related parameters	Hippo4J Server Web Add thread pool record to console
function	Provide basic functions , Including parameter dynamic 、 Runtime monitoring 、 Alarm, etc.	Extend the basic functions , Provide console interface 、 Thread pool stack view 、 Thread pool running information can be viewed in real time 、 View historical operation information 、 Thread pool configuration, cluster personalization and other functions

Use advice Select according to the situation of the company , If the basic functions can meet the requirements , choice hippo4j-core that will do ; If you want more functions , You can choose hippo4j-server.

What problem to solve

Compared with the native thread pool ,Hippo4J Solved the following problems

The parameters cannot be reasonably evaluated when the native thread pool is created ,Hippo4J The parameters can be dynamically adjusted
- When using the native thread pool , The existing parameter configuration is in case of sudden flow peak , May reject tasks frequently
- Hippo4J It provides the function of dynamically modifying parameters , Avoid restarting online applications after modifying thread pool parameters
Native thread pool no task alarm policy ,Hippo4J Provides 4 Three alarm strategies , Namely
- Activity alarm
- Queue capacity alarm
- Reject policy alarms
- Alarm for long running time
Hippo4J Provides monitoring of the running status of the native thread pool
Hippo4J Provides a view of the native thread pool runtime stack information

How the thread pool manages threads

ref 1- The core principle of thread pool + Practice of dynamically adjusting thread pool parameters | CSDN

How does the thread pool maintain its own state

About this part , See article 「 The state of the thread pool 」 chapter .

Worker Threads

Thread pool to master the state of thread and maintain the life cycle of thread , Designed the working thread in the thread pool Worker.

private final class Worker 
    extends AbastractQueueSyncChronizer 
    implements Runnable {
        ...
}

Worker This worker thread , Realized Runnable Interface , And hold a thread thread, An initialization task firstTask.thread It is used when calling the constructor ThreadFactory To create a thread , Can be used to perform tasks .firstTask Use it to save the first incoming task , This task can have or can be null

If this value is not empty , Then the thread will execute this task immediately at the beginning of startup , It also corresponds to the situation when the core thread is created
If this value is empty , Then you need to create a thread to execute the task list （workQueue） The task , That is, the creation of non core threads .

Worker The task execution model is shown in the figure below .

Thread pools need to manage the life cycle of threads , It needs to be recycled when the online program is not running for a long time . Thread pool uses a Hash Table to hold thread references , This can be done by adding references 、 Remove references to control the life cycle of threads . At this time, it is important to determine whether the thread is running .

Worker By inheritance AQS, Use AQS To realize the function of exclusive lock . No re-entry locks are used ReentrantLock, But use AQS, In order to realize the non reentrant feature to reflect the current execution state of the thread .

Worker Threads and thread creation 、 Recycling

For details of this chapter, please refer to Java The principle of thread pool and its practice in meituan business , Only outline records are made here .

Add threads
- The increase of threads is through addWorker Method , The function of this method is to add a thread , This method does not consider the stage in which the thread pool is added , This strategy of allocating threads was completed in the last step , This step only completes adding threads , And make it run , Finally, it returns whether the result is successful .
- addWorker Method has two parameters ：firstTask、core.firstTask Parameter is used to specify the first task executed by the new thread , This parameter can be empty ;core Parameter is true Indicates whether the current number of active threads is less than when a new thread is added corePoolSize,false Indicates whether the number of active threads is less than before adding a new thread maximumPoolSize
Thread recycling
- Destroy dependency of thread in thread pool JVM Automatic recycling
- The task of thread pool is to maintain a certain number of thread references according to the current state of thread pool , Prevent this part of the thread from being JVM Recycling , When the thread pool decides which threads need to be recycled , Just eliminate the references
- Worker After being created , It's going to keep polling , Then get the task to execute , Core threads can wait indefinitely for get tasks , Non core threads need to get tasks in a limited time
- When Worker Can't get task , That is, when the acquired task is empty , The cycle will end ,Worker It will actively eliminate references in its own process pool

Implementation principle of dynamic adjustment of thread pool parameters

ref 1- Thread pool parameter setting and dynamic adjustment | CSDN

Think with the following questions

What are the pain points of the existing thread pool parameter setting scheme
How dynamic updates work
What should be paid attention to in dynamic setting
How to dynamically specify queue length

Number of core threads -setCorePoolSize

ThreadPoolExecutor Provide setCorePoolSize() Method , You can adjust the number of core threads in the thread pool .

Spring Provides ThreadPoolTaskExecutor Manage thread pools , Also provided setCorePoolSize() Method , Internally, by calling ThreadPoolExecutor Of setCorePoolSize() Method , As shown in the following code .

/**
* This setting can be modified at runtime  
*  Can be dynamically adjusted at runtime 
*/
public void setCorePoolSize(int corePoolSize) {
   synchronized(this.poolSizeMonitor) {
      this.corePoolSize = corePoolSize;
      if (this.threadPoolExecutor != null) {
         this.threadPoolExecutor.setCorePoolSize(corePoolSize);
      }
   }
}

During thread pool operation , Use this method to set corePoolSize after , The thread pool will directly cover the original corePoolSize value , And based on the comparison between the current value and the original value （ Make difference , In fact, when setting the thread pool for the first time, the difference is also done , It's just the first time , The original value was zero 0 nothing more ）, Adopt different processing strategies

If the current value is less than the current number of worker threads , Explain that there are redundant worker Threads , The current idle Of worker Thread initiates interrupt request to implement recycle , Redundant worker The next time idel It's also recycled ;
For the current value is greater than the original value and the task to be executed in the current queue , The thread pool will create a new worker Thread to execute the queue task

Maximum number of threads -setMaximumPoolSize

The maximum number of threads is related to memory , and CPU Kernel independent .CPU The number of cores is related to the running efficiency of threads .

    /**
     * Sets the maximum allowed number of threads. This overrides any
     * value set in the constructor. If the new value is smaller than
     * the current value, excess existing threads will be
     * terminated when they next become idle.
     *
     * @param maximumPoolSize the new maximum
     * @throws IllegalArgumentException if the new maximum is
     *         less than or equal to zero, or
     *         less than the {@linkplain #getCorePoolSize core pool size}
     * @see #getMaximumPoolSize
     */
    public void setMaximumPoolSize(int maximumPoolSize) {
        if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
            throw new IllegalArgumentException();
        this.maximumPoolSize = maximumPoolSize;
        if (workerCountOf(ctl.get()) > maximumPoolSize)
            interruptIdleWorkers();
    }

setMaximumPoolSize() Can be used to adjust the maximum number of threads , The process is as follows

The first is the validation of parameters
And then use the value passed in , Overwrite the original value
Determine whether the worker thread is greater than the maximum number of threads . If it is greater than , The interrupt request is made to the idle thread

Adjust the queue length

Dynamically adjust thread pool parameters , It mainly adjusts three parameters , Number of core threads 、 Maximum number of threads and queue length . The first two have corresponding set Method , But the adjustment of queue length , There is no corresponding set Method .

Why not? set Methods? ？ Check the source code to find out , Queued capacity Be being final Embellished .

/** The capacity bound, or Integer.MAX_VALUE if none */
private final int capacity;

Reference resources 「 Meituan dynamic thread pool design 」, You can create a ResizableCapacityLinkedBlockIngQueue Queues （ A copy of LinkedBlockingQueue Code for , Change on this basis ）, change capacity Of final attribute , And provide the corresponding get/set Method , So as to achieve the purpose of dynamically adjusting the queue length .

matters needing attention

The number of core threads may be invalid after adjustment

During dynamic adjustment , The number of core threads may be invalid after adjustment , such as

In the initial state , The number of core threads is 2, The maximum number of threads is 5
During dynamic adjustment , Modify the number of core threads to 10
At this point, check the thread pool status , You will find that the number of core threads has indeed become 10, But the number of active threads is still 5

Combine the following ThreadPoolExecutor#getTask Source code , see if(wc > maximumPoolSize) This business , You can know , If the number of worker threads （ Number of core threads ） Greater than the maximum number of threads , The number of worker threads is reduced by one , Then return null.

therefore , The actual process here should be

Create a new worker thread worker, Then add one to the number of worker threads , Run the created worker thread worker, Start getting the task task.
The number of worker threads is greater than the maximum number of threads , Subtract the number of worker threads by one , return null, That is, they didn't get task, Clean up the task , End of the process .
So one plus one minus , So the number of working threads actually executing tasks , Nothing has changed , That's the maximum number of threads .

    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

How to solve this problem , You only need to set the number of core threads , Set the maximum number of threads at the same time , You can actually set them to the same value

In the initial state , The number of core threads is 2, The maximum number of threads is 5
During dynamic adjustment , Modify the number of core threads to 10, At the same time, adjust the maximum number of threads to 10
At this point, check the thread pool status , You will find that the number of core threads has become 10, The number of active threads has also become 10

Can core threads be recycled

During dynamic adjustment , If you manually increase the number of core threads , Do you still need to manually reduce it during the low peak period of business ？

The answer is no need for .corePoolSize The source code comments are as follows , You can find , If set allowCoreThreadTimeOut（ This value defaults to false）, The number of core threads exceeds keepAliveTime It will also be recycled automatically .

    /**
     * Core pool size is the minimum number of workers to keep alive
     * (and not allow to time out etc) unless allowCoreThreadTimeOut
     * is set, in which case the minimum is zero.
     */
    private volatile int corePoolSize;

allowCoreThreadTimeOut The relevant source code is as follows .

    /**
     * Returns true if this pool allows core threads to time out and
     * terminate if no tasks arrive within the keepAlive time, being
     * replaced if needed when new tasks arrive. When true, the same
     * keep-alive policy applying to non-core threads applies also to
     * core threads. When false (the default), core threads are never
     * terminated due to lack of incoming tasks.
     *
     * @return {@code true} if core threads are allowed to time out,
     *         else {@code false}
     *
     * @since 1.6
     */
    public boolean allowsCoreThreadTimeOut() {
        return allowCoreThreadTimeOut;
    }

    public void allowCoreThreadTimeOut(boolean value) {
        if (value && keepAliveTime <= 0)
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        if (value != allowCoreThreadTimeOut) {
            allowCoreThreadTimeOut = value;
            if (value)
                interruptIdleWorkers();
        }
    }

Thread pool warm up

Is there any thread in the thread pool after it is created ？ If not , Do you know any way to warm up the thread pool ？

After the thread pool is created , If there is no task to come , There won't be threads in it . If preheating is needed , You can call ThreadPoolExecutor The next two methods of

prestartAllCoreThreads It will warm up all threads

    /**
     * Starts all core threads, causing them to idly wait for work. This
     * overrides the default policy of starting core threads only when
     * new tasks are executed.
     *
     * @return the number of threads started
     */
    public int prestartAllCoreThreads() {
        int n = 0;
        while (addWorker(null, true))
            ++n;
        return n;
    }

prestartCoreThread Will start a core thread

    /**
     * Starts a core thread, causing it to idly wait for work. This
     * overrides the default policy of starting core threads only when
     * new tasks are executed. This method will return {@code false}
     * if all core threads have already been started.
     *
     * @return {@code true} if a thread was started
     */
    public boolean prestartCoreThread() {
        return workerCountOf(ctl.get()) < corePoolSize &&
            addWorker(null, true);
    }

原网站

版权声明
本文为[Bug Commander]所创，转载请带上原文链接，感谢
https://yzsam.com/2022/183/202207020249016199.html