Summary:How does Java implement and manage the thread pool?

This article is shared from Huawei Cloud Community “JUC Thread Pool: Detailed Explanation of ThreadPoolExecutor”, author: Notes of Brother Long.

Take the interview questions from big factories to understand

hint

Please continue with these questions, which will help you better understand the relevant knowledge points to a large extent. @pdai

  • Why is there a thread pool?
  • What are the ways in which Java implements and manages the thread pool? Please give a simple example of how to use it.
  • Why are many companies not allowed to use Executors to create thread pools? So how do you recommend using them?
  • What are the core configuration parameters of ThreadPoolExecutor? Please explain briefly
  • Which three thread pools can ThreadPoolExecutor create?
  • What happens when the queue is full and the number of workers reaches maxSize?
  • Tell me what RejectedExecutionHandler strategies ThreadPoolExecutor has? What is the default strategy?
  • Briefly talk about the task execution mechanism of the thread pool? execute –> addWorker –>runworker (getTask)
  • How are tasks submitted in the thread pool?
  • How are tasks in the thread pool closed?
  • What configuration factors need to be considered when configuring the thread pool?
  • How to monitor the status of the thread pool?

Why have a thread pool

The thread pool can uniformly allocate, tune and monitor threads:

  • Reduce resource consumption (threads are created indefinitely and then destroyed after use)
  • Improve response speed (no need to create threads)
  • Improve thread manageability

ThreadPoolExecutor example

How does Java implement and manage the thread pool?

Starting from JDK 5, the unit of work is separated from the execution mechanism. The unit of work includes Runnable and Callable, and the execution mechanism is provided by the Executor framework.

public class WorkerThread implements Runnable {
 private String command;
 public WorkerThread(String s){
 this.command=s;
 }
 @Override
 public void run() {
 System.out.println(Thread.currentThread().getName()+" Start. Command = "+command);
 processCommand();
 System.out.println(Thread.currentThread().getName()+" End.");
 }
 private void processCommand() {
 try {
 Thread.sleep(5000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 @Override
 public String toString(){
 return this.command;
 }
}
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SimpleThreadPool {
 public static void main(String[] args) {
 ExecutorService executor = Executors.newFixedThreadPool(5);
 for (int i = 0; i < 10; i++) {
 Runnable worker = new WorkerThread("" + i);
 executor.execute(worker);
 }
 executor.shutdown(); // This will make the executor accept no new threads and finish all existing threads in the queue
 while (!executor.isTerminated()) { // Wait until all threads are finish,and also you can use "executor.awaitTermination();" to wait
 }
 System.out.println("Finished all threads");
 }
}

In the program, we created a thread pool with a fixed size of five worker threads. Then assign ten jobs to the thread pool. Because the thread pool size is five, it will start five worker threads to process five jobs first, and the other jobs will be in a waiting state. Once a job is completed, the worker thread will pick it up when it is idle. Waiting for other jobs in the queue to execute.

Here is the output of the above program.

pool-1-thread-2 Start. Command = 1
pool-1-thread-4 Start. Command = 3
pool-1-thread-1 Start. Command = 0
pool-1-thread-3 Start. Command = 2
pool-1-thread-5 Start. Command = 4
pool-1-thread-4 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
pool-1-thread-3 End.
pool-1-thread-3 Start. Command = 8
pool-1-thread-2 End.
pool-1-thread-2 Start. Command = 9
pool-1-thread-1 Start. Command = 7
pool-1-thread-5 Start. Command = 6
pool-1-thread-4 Start. Command = 5
pool-1-thread-2 End.
pool-1-thread-4 End.
pool-1-thread-3 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
Finished all threads

The output shows that there are only five threads named “pool-1-thread-1” to “pool-1-thread-5” in the thread pool from beginning to end, and these five threads do not die with the completion of the work, It will always exist and is responsible for executing the tasks assigned to the thread pool until the thread pool dies.

The Executors class provides a simple ExecutorService implementation that uses ThreadPoolExecutor, but ThreadPoolExecutor provides much more than that. We can specify the number of active threads when creating a ThreadPoolExecutor instance, we can also limit the size of the thread pool and create our own implementation of RejectedExecutionHandler to handle work that cannot fit in the work queue.

Here is the implementation of our custom RejectedExecutionHandler interface.

  • RejectedExecutionHandlerImpl.java
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
 @Override
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 System.out.println(r.toString() + " is rejected");
 }
}

ThreadPoolExecutor provides methods that we can use to query the executor’s current state, thread pool size, number of active threads, and number of tasks. So I use a monitoring thread to print executor information at specific time intervals.

import java.util.concurrent.ThreadPoolExecutor;
public class MyMonitorThread implements Runnable
{
 private ThreadPoolExecutor executor;
 private int seconds;
 private boolean run=true;
 public MyMonitorThread(ThreadPoolExecutor executor, int delay)
 {
 this.executor = executor;
 this.seconds=delay;
 }
 public void shutdown(){
 this.run=false;
 }
 @Override
 public void run()
 {
 while(run){
 System.out.println(
 String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
 this.executor.getPoolSize(),
 this.executor.getCorePoolSize(),
 this.executor.getActiveCount(),
 this.executor.getCompletedTaskCount(),
 this.executor.getTaskCount(),
 this.executor.isShutdown(),
 this.executor.isTerminated()));
 try {
 Thread.sleep(seconds*1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 }
}

Here is an example thread pool implementation using ThreadPoolExecutor.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class WorkerPool {
 public static void main(String args[]) throws InterruptedException{
 //RejectedExecutionHandler implementation
 RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
 //Get the ThreadFactory implementation to use
 ThreadFactory threadFactory = Executors.defaultThreadFactory();
 //creating the ThreadPoolExecutor
 ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), threadFactory, rejectionHandler);
 //start the monitoring thread
 MyMonitorThread monitor = new MyMonitorThread(executorPool, 3);
 Thread monitorThread = new Thread(monitor);
        monitorThread.start();
 //submit work to the thread pool
 for(int i=0; i<10; i++){
            executorPool.execute(new WorkerThread("cmd"+i));
 }
 Thread.sleep(30000);
 //shut down the pool
        executorPool.shutdown();
 //shut down the monitor thread
 Thread.sleep(5000);
 monitor.shutdown();
 }
}

Note that when initializing the ThreadPoolExecutor, we keep the initial pool size at 2, the maximum pool size at 4 and the work queue size at 2. So if there are already four executing tasks and more tasks are assigned at this time, the work queue will only hold two of them (new tasks), and the others will be processed by the RejectedExecutionHandlerImpl.

The output of the above program can confirm the above point.

pool-1-thread-1 Start. Command = cmd0
pool-1-thread-4 Start. Command = cmd5
cmd6 is rejected
pool-1-thread-3 Start. Command = cmd4
pool-1-thread-2 Start. Command = cmd1
cmd7 is rejected
cmd8 is rejected
cmd9 is rejected
[monitor] [0/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-4 End.
pool-1-thread-1 End.
pool-1-thread-2 End.
pool-1-thread-3 End.
pool-1-thread-1 Start. Command = cmd3
pool-1-thread-4 Start. Command = cmd2
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-1 End.
pool-1-thread-4 End.
[monitor] [4/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true

Note the change in the number of executors’ active tasks, completed tasks, and all completed tasks. We can call the shutdown() method to end all submitted tasks and terminate the thread pool.

Detailed use of ThreadPoolExecutor

In fact, the implementation principle of the java thread pool is very simple. To put it bluntly, it is a thread collection workerSet and a blocking queue workQueue. When the user submits a task (that is, a thread) to the thread pool, the thread pool will first put the task into the workQueue. The threads in workerSet will continuously get threads from workQueue and execute them. When there is no task in the workQueue, the worker will block until there is a task in the queue, then take it out and continue execution.

Execute principle

After a task is submitted to the thread pool:

  1. First of all, whether the number of threads currently running in the thread pool is less than corePoolSize. If yes, create a new worker thread to perform the task. If all tasks are being executed, go to 2.
  2. Determine whether the BlockingQueue is full, and if not, put the thread into the BlockingQueue. Otherwise go to 3.
  3. If creating a new worker thread would cause the number of currently running threads to exceed the maximumPoolSize, it is handed over to the RejectedExecutionHandler to handle the task.

When ThreadPoolExecutor creates a new thread, update the status ctl of the thread pool through CAS.

parameter

public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 RejectedExecutionHandler handler)
  • corePoolSize The number of core threads in the thread pool. When a task is submitted, the thread pool creates a new thread to execute the task until the current number of threads is equal to corePoolSize. Even if there are other idle threads that can execute new tasks, threads will continue to be created; if The current number of threads is corePoolSize, and the tasks that continue to be submitted are saved in the blocking queue and wait to be executed; if the prestartAllCoreThreads() method of the thread pool is executed, the thread pool will create and start all core threads in advance.
  • workQueue is used to save the blocking queue of tasks waiting to be executed. The following blocking queues are provided in the JDK: For details, please refer to JUC Collection: BlockQueue Detailed Explanation
    • ArrayBlockingQueue: A bounded blocking queue based on an array structure, sorting tasks by FIFO;
    • LinkedBlockingQueue: A blocking queue based on a linked list structure, sorting tasks by FIFO, and the throughput is usually higher than that of ArrayBlockingQueue;
    • SynchronousQueue: A blocking queue that does not store elements. Each insertion operation must wait until another thread calls the removal operation, otherwise the insertion operation is always blocked, and the throughput is usually higher than LinkedBlockingQueue;
    • PriorityBlockingQueue: an unbounded blocking queue with priority;

LinkedBlockingQueue is better than ArrayBlockingQueue in inserting and deleting nodes, but both of them need to be locked during put() and take() tasks. SynchronousQueue uses a lock-free algorithm and executes according to the state of the node without using locks , whose core is Transfer.transfer().

  • maximumPoolSize The maximum number of threads allowed in the thread pool. If the current blocking queue is full and the task continues to be submitted, a new thread will be created to execute the task, provided that the current number of threads is less than the maximumPoolSize; when the blocking queue is an unbounded queue, the maximumPoolSize will not work because it cannot be submitted to the core thread pool The thread will be continuously put into the workQueue.
  • keepAliveTime The survival time when the thread is idle, that is, when the thread has no tasks to execute, the thread continues to survive; by default, this parameter is only useful when the number of threads is greater than corePoolSize, and idle threads exceeding this time will be terminated;
  • unit keepAliveTime unit
  • threadFactory Creates a factory for threads. Through a custom thread factory, a recognizable thread name can be set for each newly created thread. The default is DefaultThreadFactory
  • The saturation strategy of the handler thread pool. When the blocking queue is full and there are no idle worker threads, if the task continues to be submitted, a strategy must be adopted to process the task. The thread pool provides 4 strategies:
    • AbortPolicy: Throw an exception directly, the default policy;
    • CallerRunsPolicy: Use the thread of the caller to execute the task;
    • DiscardOldestPolicy: Discard the front task in the blocking queue and execute the current task;
    • DiscardPolicy: Discard tasks directly;

Of course, you can also implement the RejectedExecutionHandler interface according to the application scenario, and customize the saturation strategy, such as logging or persistently storing tasks that cannot be processed.

three types

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
 return new ThreadPoolExecutor(nThreads, nThreads,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());
}

After the number of threads in the thread pool reaches corePoolSize, even if the thread pool has no executable tasks, the threads will not be released.

The work queue of FixedThreadPool is an unbounded queue LinkedBlockingQueue (the queue capacity is Integer.MAX_VALUE), which will cause the following problems:

  • The number of threads in the thread pool does not exceed corePoolSize, which leads to maximumPoolSize and keepAliveTime will be useless parameters
  • Due to the use of unbounded queues, FixedThreadPool will never refuse, that is, the saturation strategy will fail

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
 return new FinalizableDelegatedExecutorService
 (new ThreadPoolExecutor(1, 1,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>()));
}

There is only one thread in the initialized thread pool. If the thread ends abnormally, a new thread will be recreated to continue executing the task. The only thread can guarantee the order execution of the submitted tasks.

Due to the use of unbounded queues, SingleThreadPool will never refuse, that is, the saturation strategy will fail

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
 60L, TimeUnit.SECONDS,
 new SynchronousQueue<Runnable>());
}

The number of threads in the thread pool can reach Integer.MAX_VALUE, which is 2147483647, and SynchronousQueue is used internally as a blocking queue; unlike the thread pool created by newFixedThreadPool, newCachedThreadPool will automatically release thread resources when there is no task to execute and the idle time of the thread exceeds keepAliveTime. When submitting a new task, if there is no idle thread, creating a new thread to execute the task will cause a certain system overhead; the execution process is slightly different from the previous two:

  • The main thread calls the offer() method of the SynchronousQueue and puts it into the task. If there is an idle thread in the thread pool trying to read the task of the SynchronousQueue at this time, that is, poll() of the SynchronousQueue is called, then the main thread hands the task to the idle thread. Otherwise do (2)
  • When the thread pool is empty or there are no idle threads, create a new thread to execute the task.
  • If the thread that has completed the task is still idle within 60s, it will be terminated. Therefore, the CachedThreadPool that has been idle for a long time will not hold any thread resources.

Close the thread pool

Traverse all the threads in the thread pool, and call the interrupt method of the thread one by one to interrupt the thread.

Shutdown method – shutdown

Set the thread state in the thread pool to SHUTDOWN state, and then interrupt all threads that are not executing tasks.

Shutdown method – shutdownNow

Set the thread state in the thread pool to STOP state, and then stop all threads that are executing or suspending tasks. Just call any of these two shutdown methods, isShutDown() returns true. When all tasks are successfully closed, isTerminated () returns true.

Detailed source code of ThreadPoolExecutor

a few key properties

//这个属性是用来存放 当前运行的worker数量以及线程池状态的
//int是32位的，这里把int的高3位拿来充当线程池状态的标志位,后29位拿来充当当前运行worker的数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//存放任务的阻塞队列
private final BlockingQueue<Runnable> workQueue;
//worker的集合,用set来存放
private final HashSet<Worker> workers = new HashSet<Worker>();
//历史达到的worker数最大值
private int largestPoolSize;
//当队列满了并且worker的数量达到maxSize的时候,执行具体的拒绝策略
private volatile RejectedExecutionHandler handler;
//超出coreSize的worker的生存时间
private volatile long keepAliveTime;
//常驻worker的数量
private volatile int corePoolSize;
//最大worker的数量,一般当workQueue满了才会用到这个参数
private volatile int maximumPoolSize;

internal state

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
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;
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

Among them, the AtomicInteger variable ctl is very powerful: use the lower 29 bits to indicate the number of threads in the thread pool, and use the upper 3 bits to indicate the running status of the thread pool:

  • RUNNING: -1 << COUNT_BITS, that is, the upper 3 bits are 111. The thread pool in this state will receive new tasks and process tasks in the blocking queue;
  • SHUTDOWN: 0 << COUNT_BITS, that is, the upper 3 bits are 000, the thread pool in this state will not receive new tasks, but will process tasks in the blocking queue;
  • STOP : 1 << COUNT_BITS, that is, the upper 3 bits are 001, threads in this state will not receive new tasks, nor will they process tasks in the blocking queue, and will interrupt running tasks;
  • TIDYING : 2 << COUNT_BITS, that is, the upper 3 bits are 010, all tasks have been terminated;
  • TERMINATED: 3 << COUNT_BITS, that is, the upper 3 bits are 011, the terminated() method has been executed

task execution

execute –> addWorker –>runworker (getTask)

The working thread of the thread pool is implemented by the Woker class. Under the guarantee of the ReentrantLock lock, the Woker instance is inserted into the HashSet, and the thread in the Woker is started. From the implementation of the construction method of the Woker class, it can be found that when the thread factory creates a thread thread, it passes in the Woker instance itself this as a parameter. When the start method is executed to start the thread thread, the essence is to execute the worker’s runWorker method. After the execution of firstTask is completed, the waiting task is obtained from the blocking queue through the getTask method. If there is no task in the queue, the getTask method will be blocked and suspended, and will not occupy cpu resources;

execute() method

ThreadPoolExecutor.execute(task) implements Executor.execute(task)

public void execute(Runnable command) {
 if (command == null)
 throw new NullPointerException();
 /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
 int c = ctl.get();
 if (workerCountOf(c) < corePoolSize) { 
 //workerCountOf获取线程池的当前线程数；小于corePoolSize，执行addWorker创建新线程执行command任务
 if (addWorker(command, true))
 return;
        c = ctl.get();
 }
 // double check: c, recheck
 // 线程池处于RUNNING状态，把提交的任务成功放入阻塞队列中
 if (isRunning(c) && workQueue.offer(command)) {
 int recheck = ctl.get();
 // recheck and if necessary 回滚到入队操作前，即倘若线程池shutdown状态，就remove(command)
 //如果线程池没有RUNNING，成功从阻塞队列中删除任务，执行reject方法处理任务
 if (! isRunning(recheck) && remove(command))
 reject(command);
 //线程池处于running状态，但是没有线程，则创建线程
 else if (workerCountOf(recheck) == 0)
 addWorker(null, false);
 }
 // 往线程池中创建新的线程失败，则reject任务
 else if (!addWorker(command, false))
 reject(command);
}
  • Why do you need to double check the status of the thread pool?

In a multi-threaded environment, the status of the thread pool is changing all the time, and ctl.get() is a non-atomic operation. It is very likely that the status of the thread pool will change just after the status of the thread pool is obtained. Judging whether to add the command to the workque is the state before the thread pool. If there is no double check, if the thread pool is in a non-running state (which is likely to happen in a multi-threaded environment), then the command will never be executed.

addWorker method

From the implementation of the execute method, we can see that addWorker is mainly responsible for creating new threads and executing tasks. When the thread pool creates new threads to perform tasks, it needs to acquire a global lock:

private final ReentrantLock mainLock = new ReentrantLock();

private boolean addWorker(Runnable firstTask, boolean core) {
 // CAS更新线程池数量
    retry:
 for (;;) {
 int c = ctl.get();
 int rs = runStateOf(c);
 // Check if queue empty only if necessary.
 if (rs >= SHUTDOWN &&
 ! (rs == SHUTDOWN &&
                firstTask == null &&
 ! workQueue.isEmpty()))
 return false;
 for (;;) {
 int wc = workerCountOf(c);
 if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
 return false;
 if (compareAndIncrementWorkerCount(c))
 break retry;
            c = ctl.get(); // Re-read ctl
 if (runStateOf(c) != rs)
 continue retry;
 // else CAS failed due to workerCount change; retry inner loop
 }
 }
 boolean workerStarted = false;
 boolean workerAdded = false;
 Worker w = null;
 try {
        w = new Worker(firstTask);
 final Thread t = w.thread;
 if (t != null) {
 // 线程池重入锁
 final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
 try {
 // Recheck while holding lock.
 // Back out on ThreadFactory failure or if
 // shut down before lock acquired.
 int rs = runStateOf(ctl.get());
 if (rs < SHUTDOWN ||
 (rs == SHUTDOWN && firstTask == null)) {
 if (t.isAlive()) // precheck that t is startable
 throw new IllegalThreadStateException();
                    workers.add(w);
 int s = workers.size();
 if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
 }
 } finally {
                mainLock.unlock();
 }
 if (workerAdded) {
                t.start(); // 线程启动，执行任务(Worker.thread(firstTask).start());
                workerStarted = true;
 }
 }
 } finally {
 if (! workerStarted)
 addWorkerFailed(w);
 }
 return workerStarted;
}

The runworker method of the Worker class

 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
 Worker(Runnable firstTask) {
 setState(-1); // inhibit interrupts until runWorker
 this.firstTask = firstTask;
 this.thread = getThreadFactory().newThread(this); // 创建线程
 }
 /** Delegates main run loop to outer runWorker  */
 public void run() {
 runWorker(this);
 }
 // ...
 }
  • Inheriting the AQS class, it is convenient to realize the suspension operation of the worker thread;
  • Implement the Runnable interface, which can execute itself as a task in the worker thread;
  • The currently submitted task firstTask is passed as a parameter to the constructor of Worker;

Some properties also have constructors:

//运行的线程,前面addWorker方法中就是直接通过启动这个线程来启动这个worker
final Thread thread;
//当一个worker刚创建的时候,就先尝试执行这个任务
Runnable firstTask;
//记录完成任务的数量
volatile long completedTasks;
Worker(Runnable firstTask) {
 setState(-1); // inhibit interrupts until runWorker
 this.firstTask = firstTask;
 //创建一个Thread,将自己设置给他,后面这个thread启动的时候,也就是执行worker的run方法
 this.thread = getThreadFactory().newThread(this);
}

The runWorker method is the core of the thread pool:

  • After the thread is started, the lock is released through the unlock method, and the state of AQS is set to 0, indicating that the operation can be interrupted;
  • Worker executes firstTask or gets tasks from workQueue:
    • Perform locking operations to ensure that threads are not interrupted by other threads (unless the thread pool is interrupted)
    • Check the thread pool status, if the thread pool is interrupted, the current thread will be interrupted.
    • execute beforeExecute
    • The run method to execute the task
    • Execute the afterExecute method
    • Unlock operation

Get the waiting task from the blocking queue through the getTask method. If there is no task in the queue, the getTask method will be blocked and suspended, and will not occupy cpu resources;

final void runWorker(Worker w) {
 Thread wt = Thread.currentThread();
 Runnable task = w.firstTask;
 w.firstTask = null;
 w.unlock(); // allow interrupts
 boolean completedAbruptly = true;
 try {
 // 先执行firstTask，再从workerQueue中取task(getTask())
 while (task != null || (task = getTask()) != null) {
 w.lock();
 // If pool is stopping, ensure thread is interrupted;
 // if not, ensure thread is not interrupted.  This
 // requires a recheck in second case to deal with
 // shutdownNow race while clearing interrupt
 if ((runStateAtLeast(ctl.get(), STOP) ||
 (Thread.interrupted() &&
 runStateAtLeast(ctl.get(), STOP))) &&
 !wt.isInterrupted())
 wt.interrupt();
 try {
 beforeExecute(wt, task);
 Throwable thrown = null;
 try {
 task.run();
 } catch (RuntimeException x) {
                    thrown = x; throw x;
 } catch (Error x) {
                    thrown = x; throw x;
 } catch (Throwable x) {
                    thrown = x; throw new Error(x);
 } finally {
 afterExecute(task, thrown);
 }
 } finally {
                task = null;
 w.completedTasks++;
 w.unlock();
 }
 }
        completedAbruptly = false;
 } finally {
 processWorkerExit(w, completedAbruptly);
 }
}

getTask method

Let’s take a look at the getTask() method, which involves the use of keepAliveTime. From this method, we can see how the thread pool destroys the part of the worker that exceeds the corePoolSize.

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

Note that this piece of code is the key to keepAliveTime working:

boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();

allowCoreThreadTimeOut is false, the thread will not be destroyed even if it is idle; if it is true, it will be destroyed if it is still idle within keepAliveTime.

If the thread is allowed to wait idle without being destroyed timed == false, workQueue.take task: If the blocking queue is empty, the current thread will be suspended and wait; when a task is added in the queue, the thread will be woken up, and the take method returns the task, and execute;

If the thread does not allow endless idle timed == true, workQueue.poll task: If the blocking queue still has no tasks within the keepAliveTime time, return null;

submission of tasks

  1. submit task, wait for the thread pool to execute
  2. When the get method of the FutureTask class is executed, the main thread will be encapsulated into a WaitNode node and stored in the waiters list, and will be blocked to wait for the running result;
  3. After the execution of the FutureTask task is completed, set the waitNode corresponding to the waiters to null through UNSAFE, and wake up the main thread through the unpark method of the LockSupport class;
public class Test{
 public static void main(String[] args) {
 ExecutorService es = Executors.newCachedThreadPool();
 Future<String> future = es.submit(new Callable<String>() {
 @Override
 public String call() throws Exception {
 try {
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "future result";
 }
 });
 try {
 String result = future.get();
 System.out.println(result);
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
}

In actual business scenarios, Future and Callable basically appear in pairs, Callable is responsible for generating results, and Future is responsible for obtaining results.

  1. The Callable interface is similar to Runnable, except that Runnable does not return a value.
  2. In addition to returning the normal results of the Callable task, if an exception occurs, the exception will also be returned, that is, the Future can get various results of the asynchronous execution task;
  3. The Future.get method will cause the main thread to block until the Callable task is executed;

submit method

AbstractExecutorService.submit() implements ExecutorService.submit() to obtain the return value after execution, and ThreadPoolExecutor is a subclass of AbstractExecutorService.submit(), so the submit method is also a method of ThreadPoolExecutor`.

// submit()在ExecutorService中的定义
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
// submit方法在AbstractExecutorService中的实现
public Future<?> submit(Runnable task) {
 if (task == null) throw new NullPointerException();
 // 通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。
 RunnableFuture<Void> ftask = newTaskFor(task, null);
 execute(ftask);
 return ftask;
}

The Callable task submitted through the submit method will be encapsulated into a FutureTask object. Submit the FutureTask to the thread pool through the Executor.execute method to wait for execution, and the final execution is the run method of the FutureTask;

FutureTask object

public class FutureTask implements RunnableFuture can submit FutureTask to the thread pool to be executed (executed by the run method of FutureTask)

/* The run state of this task, initially NEW. 
    * ...
    * Possible state transitions:
    * NEW -> COMPLETING -> NORMAL
    * NEW -> COMPLETING -> EXCEPTIONAL
    * NEW -> CANCELLED
    * NEW -> INTERRUPTING -> INTERRUPTED
    */
private volatile int state;
private static final int NEW          = 0;
private static final int COMPLETING   = 1;
private static final int NORMAL       = 2;
private static final int EXCEPTIONAL  = 3;
private static final int CANCELLED    = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED  = 6;

The modification of the internal state is modified by sun.misc.Unsafe

public V get() throws InterruptedException, ExecutionException {
 int s = state;
 if (s <= COMPLETING)
        s = awaitDone(false, 0L);
 return report(s);
}

The main thread is blocked internally through the awaitDone method, and the specific implementation is as follows:

private int awaitDone(boolean timed, long nanos)
 throws InterruptedException {
 final long deadline = timed ? System.nanoTime() + nanos : 0L;
 WaitNode q = null;
 boolean queued = false;
 for (;;) {
 if (Thread.interrupted()) {
 removeWaiter(q);
 throw new InterruptedException();
 }
 int s = state;
 if (s > COMPLETING) {
 if (q != null)
 q.thread = null;
 return s;
 }
 else if (s == COMPLETING) // cannot time out yet
 Thread.yield();
 else if (q == null)
            q = new WaitNode();
 else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);
 else if (timed) {
            nanos = deadline - System.nanoTime();
 if (nanos <= 0L) {
 removeWaiter(q);
 return state;
 }
 LockSupport.parkNanos(this, nanos);
 }
 else
 LockSupport.park(this);
 }
}

If the main thread is interrupted, an interrupt exception is thrown;

  1. Determine the current state of the FutureTask, if it is greater than COMPLETING, it means that the task has been executed, and return directly;
  2. If the current state is equal to COMPLETING, it means that the task has been executed. At this time, the main thread only needs to yield cpu resources through the yield method, and wait for the state to become NORMAL;
  3. Encapsulate the current thread through the WaitNode class, and add it to the waiters list through UNSAFE;
  4. Finally, the thread is suspended through the park or parkNanos of LockSupport;

run method

public void run() {
 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
 return;
 try {
 Callable<V> c = callable;
 if (c != null && state == NEW) {
 V result;
 boolean ran;
 try {
                result = c.call();
                ran = true;
 } catch (Throwable ex) {
                result = null;
                ran = false;
 setException(ex);
 }
 if (ran)
 set(result);
 }
 } finally {
 // runner must be non-null until state is settled to
 // prevent concurrent calls to run()
        runner = null;
 // state must be re-read after nulling runner to prevent
 // leaked interrupts
 int s = state;
 if (s >= INTERRUPTING)
 handlePossibleCancellationInterrupt(s);
 }
}

The FutureTask.run method is executed in the thread pool, not the main thread

  1. By executing the call method of the Callable task;
  2. If the call is executed successfully, the result is saved through the set method;
  3. If there is an exception in call execution, save the exception through setException;

task closure

The shutdown method will set the state of the thread pool to SHUTDOWN. After the thread pool enters this state, it will refuse to accept tasks, and then all remaining tasks will be executed.

public void shutdown() {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //检查是否可以关闭线程
 checkShutdownAccess();
 //设置线程池状态
 advanceRunState(SHUTDOWN);
 //尝试中断worker
 interruptIdleWorkers();
 //预留方法,留给子类实现
 onShutdown(); // hook for ScheduledThreadPoolExecutor
 } finally {
        mainLock.unlock();
 }
 tryTerminate();
}
private void interruptIdleWorkers() {
 interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //遍历所有的worker
 for (Worker w : workers) {
 Thread t = w.thread;
 //先尝试调用w.tryLock(),如果获取到锁,就说明worker是空闲的,就可以直接中断它
 //注意的是,worker自己本身实现了AQS同步框架,然后实现的类似锁的功能
 //它实现的锁是不可重入的,所以如果worker在执行任务的时候,会先进行加锁,这里tryLock()就会返回false
 if (!t.isInterrupted() && w.tryLock()) {
 try {
 t.interrupt();
 } catch (SecurityException ignore) {
 } finally {
 w.unlock();
 }
 }
 if (onlyOne)
 break;
 }
 } finally {
        mainLock.unlock();
 }
}

ShutdownNow does a relatively good job. It first sets the thread pool status to STOP, and then rejects all submitted tasks. Finally, interrupt the left and right running workers, and then clear the task queue.

public List<Runnable> shutdownNow() {
 List<Runnable> tasks;
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 checkShutdownAccess();
 //检测权限
 advanceRunState(STOP);
 //中断所有的worker
 interruptWorkers();
 //清空任务队列
        tasks = drainQueue();
 } finally {
        mainLock.unlock();
 }
 tryTerminate();
 return tasks;
}
private void interruptWorkers() {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //遍历所有worker，然后调用中断方法
 for (Worker w : workers)
 w.interruptIfStarted();
 } finally {
        mainLock.unlock();
 }
}

deeper understanding

Why is the thread pool not allowed to use Executors to create? What is the recommended way?

Thread pools are not allowed to be created using Executors, but through ThreadPoolExecutor. This processing method allows students who write to be more clear about the running rules of the thread pool and avoid the risk of resource exhaustion. Description: The disadvantages of each method of Executors:

  • newFixedThreadPool and newSingleThreadExecutor: The main problem is that the accumulated request processing queue may consume very large memory, or even OOM.
  • newCachedThreadPool and newScheduledThreadPool: The main problem is that the maximum number of threads is Integer.MAX_VALUE, which may create a very large number of threads, or even OOM.

Recommended way 1

First introduced: commons-lang3 package

ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
 new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());

Recommended way 2

First introduced: com.google.guava package

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();
//Common Thread Pool
ExecutorService pool = new ThreadPoolExecutor(5, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
// excute
pool.execute(()-> System.out.println(Thread.currentThread().getName()));
 //gracefully shutdown
pool.shutdown();

Recommended way 3

Spring configuration thread pool method: custom thread factory bean needs to implement ThreadFactory, you can refer to other default implementation classes of this interface, and use it to directly inject the bean and call the execute(Runnable task) method

 <bean id="userThreadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
 <property name="corePoolSize" value="10" />
 <property name="maxPoolSize" value="100" />
 <property name="queueCapacity" value="2000" />
 <property name="threadFactory" value= threadFactory />
 <property name="rejectedExecutionHandler">
 <ref local="rejectedExecutionHandler" />
 </property>
 </bean>
    //in code
    userThreadPool.execute(thread);

Factors to consider when configuring the thread pool

From the four perspectives of task priority, task execution time, task nature (CPU-intensive/IO-intensive), and task dependencies. And use bounded work queues as close as possible.

Tasks of different nature can be processed separately using thread pools of different sizes:

  • CPU intensive: as few threads as possible, Ncpu+1
  • IO-intensive: as many threads as possible, Ncpu*2, such as database connection pool
  • Hybrid: CPU-intensive tasks and IO-intensive tasks have little difference in execution time, and are split into two thread pools; otherwise, there is no need to split.

Monitor the state of the thread pool

The following methods of ThreadPoolExecutor can be used:

  • getTaskCount() Returns the approximate total number of tasks that have ever been scheduled for execution.
  • getCompletedTaskCount() Returns the approximate total number of tasks that have completed execution. Returns less results than getTaskCount().
  • getLargestPoolSize() Returns the largest number of threads that have ever simultaneously been in the pool. The returned result is less than or equal to maximumPoolSize
  • getPoolSize() Returns the current number of threads in the pool.
  • getActiveCount() Returns the approximate number of threads that are actively executing tasks.

reference article

  • The Art of Java Concurrent Programming
  • https://www.jianshu.com/p/87bff5cc8d8c
  • https://blog.csdn.net/programmer_at/article/details/79799267
  • https://blog.csdn.net/u013332124/article/details/79587436
  • https://www.journaldev.com/1069/threadpoolexecutor-java-thread-pool-example-executorservice

Click to follow and learn about Huawei Cloud’s fresh technologies for the first time~

#Detailed #explanation #Thread #Pool #Executor #words #HUAWEI #CLOUD #Developer #Alliances #personal #space #News Fast Delivery

Leave a Comment

Your email address will not be published. Required fields are marked *