Shulou(Shulou.com)06/01 Report--

How to realize thread reuse in Java thread pool? aiming at this problem, this article introduces the corresponding analysis and solution in detail, hoping to help more partners who want to solve this problem to find a more simple and feasible method.

MySQL%22%7D%2C%7B%22label%22%3A%22Redis%22%7D%2C%7B%22label%22%3A%22Elasticsearch%22%7D%5D "data-w=" 286 "data-ratio=" 1.5664335664335665 "data-parentclass=" appmsg_search_iframe_wrp ">

A few days ago, a group friend in the technology group asked a question about the thread pool, as shown in the figure:

About thread pool, you can take a look at this article first: why does Alibaba's Java development manual force thread pools not to be created with Executors?

So let's discuss this problem with you. In the thread pool, threads will read tasks from workQueue to execute. The smallest execution unit is that Worker,Worker implements the Runnable interface and rewrites the run method. This run method allows each thread to execute a loop. In this loop code, to determine whether there is a task to be executed, if so, to execute the task directly, so the number of threads will not increase.

The following is the overall flowchart of thread pool creation threads:

The state of the thread pool is first determined, that is, whether it is running or not, and if the thread is not running, it will be rejected. Next, it will determine whether the number of threads is less than the number of core threads. If it is less than the number of core threads, it will create a new worker thread and execute the task. With the increase of the task, the number of threads will gradually increase to the number of core threads. If there is a task submission at this time, it will determine whether the blocking queue workQueue is full. If it is not full, it will put the task into the blocking queue and wait for the worker thread to get and execute. If the task is submitted very much, it will determine whether the blocking queue is full or not. When the blocking queue reaches the upper limit, it will determine whether the number of threads is less than the maximum number of threads maximumPoolSize. If it is less than the maximum number of threads, the thread pool will add worker threads and execute tasks. If there are still a large number of tasks submitted, so that the number of threads is equal to the maximum number of threads, if there are still tasks submitted at this time, they will be rejected.

Now that we have a general understanding of this process, let's take a look at how the source code is implemented.

Thread pool task submission in terms of the submit method, the submit method is defined by the AbstractExecutorService abstract class and does two main things:

Convert both Runnable and Callable to FutureTask and use the execute method to execute FutureTask

The execute method is a method in ThreadPoolExecutor. The source code is as follows:

Public void execute (Runnable command) {

/ / if the task is empty, NPE is thrown, and the empty task cannot be executed.

If (command = = null) {

Throw new NullPointerException ()

}

Int c = ctl.get ()

/ / if the number of worker threads is less than the number of core threads, create a new thread and take the current task command as the first task of this thread

If (workerCountOf (c) < corePoolSize) {

If (addWorker (command, true)) {

Return

}

C = ctl.get ()

}

/ * *

* so far, there are the following two situations:

* 1. The current number of worker threads is greater than or equal to the number of core threads

* 2. Failed to create new thread

* an attempt will be made to add the task to the blocking queue workQueue

, /

/ / if the thread pool is in RUNNING state, add the task to the blocking queue workQueue

If (isRunning (c) & & workQueue.offer (command)) {

/ / check the thread pool flag again

Int recheck = ctl.get ()

/ / if the thread pool is no longer in the RUNNING state, remove the queued task and execute the reject policy

If (! isRunning (recheck) & & remove (command)) {

/ / failed to add task to blocking queue, execute reject policy

Reject (command)

}

/ / if the thread pool is still RUNNING and the number of threads is 0, start a new thread

Else if (workerCountOf (recheck) = = 0) {

AddWorker (null, false)

}

}

/ * *

* so far, there are the following two situations:

* 1. The thread pool is not running and the thread pool no longer accepts new threads

* 2. The thread is running, but the blocking queue is full and cannot join the blocking queue

* at this point, an attempt is made to create a new worker thread based on the maximum number of threads

, /

Else if (! addWorker (command, false)) {

/ / Task failed to enter thread pool, execute reject policy

Reject (command)

}

}

You can see that the core method in the execute method is addWorker. Before you look at the addWorker method, take a look at the initialization method of Worker:

Worker (Runnable firstTask) {

/ / the lock state of each task is initialized to-1, so that the worker thread forbids interruptions before running

SetState (- 1)

This.firstTask = firstTask

/ / run Worker as a task for thread

This.thread = getThreadFactory () .newThread (this)

}

The current Worker is input as the constructor of the thread when the Worker is initialized, and the following code can be found in the addWorker method:

Final Thread t = w.thread

/ / if you successfully add Worker, you can start Worker

If (workerAdded) {

T.start ()

WorkerStarted = true

}

This code successfully adds worker, calls the start method to start the thread, Thread t = w. Thread; at this time w is a reference to Worker, then t.start (); what is actually executed is the run method of Worker.

The runWorker method is called in the run method of Worker. The simplified runWorker source code is as follows:

Final void runWorker (Worker w) {

Runnable task = w.firstTask

While (task! = null | | (task = getTask ())! = null) {

Try {

Task.run ()

} finally {

Task = null

}

}

}

This while loop has a getTask method. The main function of getTask is to block tasks taken from the queue. If there is a task in the queue, it can be taken out and executed. If there is no task in the queue, the thread will block until there is a task (or timeout blocking). The timing diagram of the getTask method is as follows:

The key to thread reuse is part 1.6 and 1.7, and the source code is as follows:

Runnable r = timed? WorkQueue.poll (keepAliveTime, TimeUnit.NANOSECONDS): workQueue.take ()

Use the poll or take method of the queue to get data from the queue. According to the characteristics of the queue, there are tasks in the queue that can be returned, and no tasks in the queue will block.

The thread reuse of thread pool is to execute the task by fetching the firstTask of Worker or through the getTask method from workQueue, and directly calling the run method of Runnable to execute the task, thus ensuring that each thread is always in a loop, repeatedly getting the task, and then executing the task, thus realizing thread reuse.

This paper mainly analyzes how to realize thread reuse in Java thread pool from the point of view of source code. This is the answer to the question about how to achieve thread reuse in the Java thread pool. I hope the above content can be of some help to you. If you still have a lot of doubts to be solved, you can follow the industry information channel for more related knowledge.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.