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

How to analyze the complete ThreadPoolExecutor, in view of 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.

ThreadPoolExecutor source code parsing

Today, in order to make a document for a friend, parse ThreadPoolExecutor from the source level. Then parse in the form of comments directly on the source code.

1. Explanation of commonly used variables / 1. `ctl` can be regarded as a number of int type. The higher 3 bits represent the thread pool status and the lower 29 bits indicate the number of worker.

Private final AtomicInteger ctl = new AtomicInteger (ctlOf (RUNNING, 0))

/ / 2. `COUNT_ BITS`, `Integer.SIZE` is 32, so `COUNT_ BITS` is 29

Private static final int COUNT_BITS = Integer.SIZE-3

/ / 3. `CAPACITY`, the maximum number of threads allowed in the thread pool. 1 moves 29 bits to the left, then minus 1, which is 2 ^ 29-1.

Private static final int CAPACITY = (1 SHUTDOWN) | |

/ (rs = = SHUTDOWN & & firstTask! = null) | |

/ / rs = = SHUTDOWN & & workQueue.isEmpty ()

/ / 1. If the thread pool state is greater than SHUTDOWN, return false directly

/ / 2. Thread pool status is equal to SHUTDOWN, and firstTask is not null, so false is returned directly.

/ / 3. Thread pool status is equal to SHUTDOWN, and the queue is empty. Return false directly.

/ / Check if queue empty only if necessary.

If (rs > = SHUTDOWN & &

! (rs = = SHUTDOWN & &

FirstTask = = null & &

! WorkQueue.isEmpty ())

Return false

/ / Inner spin

For (;;) {

Int wc = workerCountOf (c)

/ / if the number of worker exceeds the capacity, return false directly

If (wc > = CAPACITY | |

Wc > = (core? CorePoolSize: maximumPoolSize))

Return false

/ / increase the number of worker by using CAS.

/ / if the increase is successful, jump directly out of the outer loop and enter the second part

If (compareAndIncrementWorkerCount (c)

Break retry

C = ctl.get (); / / Re-read ctl

/ / the state of the thread pool changes and spins the outer loop

If (runStateOf (c)! = rs)

Continue retry

/ / in other cases, you can spin directly in the inner loop.

/ / 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

/ / the addition of worker must be serial, so it needs to be locked

MainLock.lock ()

Try {

/ / Recheck while holding lock.

/ / Back out on ThreadFactory failure or if

/ / shut down before lock acquired.

/ / Thread pool status needs to be re-checked here

Int rs = runStateOf (ctl.get ())

If (rs

< SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // worker已经调用过了start()方法，则不再创建worker if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // worker创建并添加到workers成功 workers.add(w); // 更新`largestPoolSize`变量 int s = workers.size(); if (s >

LargestPoolSize)

LargestPoolSize = s

WorkerAdded = true

}

} finally {

MainLock.unlock ()

}

/ / start the worker thread

If (workerAdded) {

T.start ()

WorkerStarted = true

}

}

} finally {

/ / worker thread startup failed, indicating that the thread pool status has changed (the shutdown operation is performed), and shutdown-related operations are required

If (! WorkerStarted)

AddWorkerFailed (w)

}

Return workerStarted

} 5. Thread pool worker task unit private final class Worker

Extends AbstractQueuedSynchronizer

Implements Runnable

{

/ * *

* This class will never be serialized, but we provide a

* serialVersionUID to suppress a javac warning.

, /

Private static final long serialVersionUID = 6138294804551838833L

/ * Thread this worker is running in. Null if factory fails. , /

Final Thread thread

/ * Initial task to run. Possibly null. , /

Runnable firstTask

/ * * Per-thread task counter * /

Volatile long completedTasks

/ * *

* Creates with given first task and thread from ThreadFactory.

* @ param firstTask the first task (null if none)

, /

Worker (Runnable firstTask) {

SetState (- 1); / / inhibit interrupts until runWorker

This.firstTask = firstTask

/ / here is the key to Worker, using a thread factory to create a thread. The parameter passed in is the current worker

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

}

/ * * Delegates main run loop to outer runWorker * /

Public void run () {

RunWorker (this)

}

/ / omit the code.

} 6. Core thread execution logic-runworkerfinal void runWorker (Worker w) {

Thread wt = Thread.currentThread ()

Runnable task = w.firstTask

W.firstTask = null

/ / unlock () is called so that the outside can interrupt

W.unlock (); / / allow interrupts

/ / this variable is used to determine whether to enter the spin (while cycle).

Boolean completedAbruptly = true

Try {

/ / here is spin

/ / 1. If firstTask is not null, execute firstTask

/ / 2. If firstTask is null, call getTask () to get the task from the queue.

/ / 3. The characteristic of the blocking queue is that when the queue is empty, the current thread will be blocked to wait

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

/ / worker is locked here to achieve the following purpose

/ / 1. Reduce lock range and improve performance

/ / 2. Ensure that the tasks performed by each worker are serial

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 the thread pool is stopping, the current thread is interrupted

If ((runStateAtLeast (ctl.get (), STOP)) | |

(Thread.interrupted () & &

RunStateAtLeast (ctl.get (), STOP)) & &

! wt.isInterrupted ()

Wt.interrupt ()

/ / execute the task, and expand its function by `beforeExecute () `and `afterExecute ()` before and after execution.

/ / these two methods are implemented empty in the current class.

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 {

/ / help gc

Task = null

/ / add one to the number of tasks completed

W.completedTasksgiving +

W.unlock ()

}

}

CompletedAbruptly = false

} finally {

/ / the spin operation is exited, indicating that the thread pool is ending

ProcessWorkerExit (w, completedAbruptly)

}

}

The answer to how to analyze the complete ThreadPoolExecutor question is shared here. 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 to learn more about it.

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.