How to use Java thread pool execute() method-javaTutorial-php.cn

First understand what the thread pool does

* Thread pools address two different problems: they usually
* provide improved performance when executing large numbers of
* asynchronous tasks, due to reduced per-task invocation overhead,
* and they provide a means of bounding and managing the resources,
* including threads, consumed when executing a collection of tasks.
* Each {@code ThreadPoolExecutor} also maintains some basic
* statistics, such as the number of completed tasks.

Copy after login

The thread pool handles two different problems. The thread pool gives better performance to a large number of asynchronous tasks by reducing the overhead before the thread is formally called. At the same time, a series of methods for binding management task threads are given. Each thread pool contains some basic information, such as the number of tasks completed internally.

Let’s first look at a series of status representations of the ThreadPoolExecutor class.

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

Copy after login

ctl as the AtomicInteger class stores two kinds of information in the class, in which the high 3 bits save the status of the thread pool. 29 bits are used to save the number of Woker class threads in the thread pool at this time (it can be seen that the maximum acceptable number of threads in the thread pool is about 500 million). It can be seen that the runStateOf() and workerCountOf() methods provided respectively provide methods to view the thread status and number of threads.

This class gives a total of five states

Let us look at the comments given by the author

*   RUNNING:  Accept new tasks and process queued tasks
*   SHUTDOWN: Don&#39;t accept new tasks, but process queued tasks
*   STOP:     Don&#39;t accept new tasks, don&#39;t process queued tasks,
*             and interrupt in-progress tasks
*   TIDYING:  All tasks have terminated, workerCount is zero,
*             the thread transitioning to state TIDYING
*             will run the terminated() hook method
*   TERMINATED: terminated() has completed

Copy after login

RUNNINGState It can accept new incoming tasks and also execute tasks in the queue.
SHUTDOWN The status will no longer accept new tasks, but tasks in the queue will still be processed.
STOP Based on the previous status, the characters in the queue will not be processed, and the tasks being executed will be directly interrupted.
TIDYINGThe status is based on the previous one, all tasks have been terminated, and the Worker threads in the pool have all been 0, that is, the stop status has been cleared after all The worker thread will then enter this state. At the same time, in the shutdown state, it will directly enter this stage after the queue is empty and the worker thread has been cleaned up. In this stage, the terminated() method will be executed cyclically.
TERMINATED The state is the last state. Based on the previous state, the terminated() method has also been executed, and then it will enter this state from the previous state, which means The thread pool has completely stopped.

Since the status of the thread pool is saved through AtomicInteger, the current thread status can be obtained simply by comparison.

private final BlockingQueue<Runnable> workQueue; 
private final ReentrantLock mainLock = new ReentrantLock(); 
private final HashSet<Worker> workers = new HashSet<Worker>(); 
private final Condition termination = mainLock.newCondition(); 
private int largestPoolSize; 
private long completedTaskCount; 
private volatile ThreadFactory threadFactory; 
private volatile RejectedExecutionHandler handler; 
private volatile long keepAliveTime; 
private volatile boolean allowCoreThreadTimeOut; 
private volatile int corePoolSize; 
private volatile int maximumPoolSize;

Copy after login

Next are several variables related to worker threads in the thread pool

corePoolSizeIndicates the minimum number of worker threads allowed to survive in the thread pool , but it is worth noting that if allowCoreThreadTimeOut is set to true (default false), the survival time of each thread is only keepAliveTime. That is to say, when allowCoreThreadTimeOut is true, the minimum number of working threads in the thread pool is 0; maximumPoolSize represents the thread pool The maximum number of worker threads.
keepAliveTime is the longest waiting time for each worker thread when waiting for work when the number of worker threads in the thread pool is greater than corePoolSize.
workQueueAs a task waiting queue for the thread pool, this will be explained in detail in the next execute().
#WorkersAs a container for storing worker threads in the thread pool.
largestPoolSize is used to record the maximum number of worker threads that have ever existed in the thread pool.
completedTaskCount is used to record the total number of tasks completed by the thread pool.
HandlerAs a rejection strategy in the thread pool when the task cannot be accepted, we can implement our own rejection strategy on the premise of implementing the RejectedExecutionHandler interface. The following is the default rejection policy of the thread pool.

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
                                         " rejected from " +
                                         e.toString());
}

Copy after login

threadFactory is used as the factory class for thread pool production threads

The following is the thread pool's default thread factory production thread method

public Thread newThread(Runnable r) {
    Thread t = new Thread(group, r,
                          namePrefix + threadNumber.getAndIncrement(),
                          0);
    if (t.isDaemon())
        t.setDaemon(false);
    if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
    return t;
}

Copy after login

We can first look at the execute() method we call most often

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
 
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

Copy after login

The internal calling logic of execute() is very clear.

If the number of worker threads in the current thread pool is less than corePoolSize, call addWoker() directly to add worker threads.

The following is the specific method of addWorker()

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        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;
        }
    }
 
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);
 
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive())                         
           throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

Copy after login

This method is relatively long, but the overall logic is still clear.

First determine the status of the current thread pool. If the status is not shutdown or running, or it is shutdown but the work queue is empty, then it will directly return to the failure to add work at this time. The next step is to judge the number of thread pool threads. Based on the core value at the time of call, whether it is judged by corePoolSize or maximumPoolSize.

After confirming the status of the thread pool and the number of worker threads in the thread pool, you can actually start adding worker threads.

Create a new worker class (internal class of thread pool, specific working thread), pass the specific thread to be executed as a parameter in the construction method, and then add it to the working thread container of the thread pool workers, and updates the maximum number of worker threads, and finally calls the start() method of the worker thread to complete the creation and startup of the worker thread.

Let's go back to the execute() method. If the number of threads we have at the beginning is greater than corePoolSize, or there is a problem when we call the addworker() method and the number of worker threads fails to be added, then we The next logic will continue to be executed.

在判断完毕线程池的状态后，则会将任务通过workQueue.offer())方法试图加进任务队列。Offer()方法的具体实现会根据在线程池构造方法中选取的任务队列种类而产生变化。

但是如果成功加入了任务队列，仍旧需要注意判断如果线程池的状态如果已经不是running那么会拒绝执行这一任务并执行相应的拒绝策略。在最后需要记得成功加入队列成功后如果线程池中如果已经没有了工作线程，需要重新建立一个工作线程去执行仍旧在任务队列中等待执行的任务。

如果在之前的前提下加入任务队列也失败了（比如任务队列已满），则会在不超过线程池最大线程数量的前提下建立一个工作线程来处理。

如果在最后的建立工作线程也失败了，那么我们只有很遗憾的执行任务的拒绝策略了。

在之前的过程中我们建立了工作线程Worker()类，那么我们现在看看worker类的内部实现，也可以说是线程池的核心部分。

Worker类作为线程池的内部类

接下来是Worker()类的成员

final Thread thread;
 
Runnable firstTask;
 
volatile long completedTasks;

Copy after login

thread作为worker的工作线程空间，由线程池中所设置的线程工厂生成。

firstTask则是worker在构造方法中所接受到的所要执行的任务。
completedTasks作为该worker类所执行完毕的任务总数。

接下来我们可以看最重要的，也就是我们之前建立完Worker类之后立马调用的run()方法了

public void run() {
    runWorker(this);
}

Copy after login

run()方法实现的很简单

我们可以继续追踪下去

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); 
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            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);
    }
}

Copy after login

如果这个worker还没有执行过在构造方法就传入的任务，那么在这个方法中，会直接执行这一任务，如果没有，则会尝试去从任务队列当中去取的新的任务。

但是在真正调用任务之前，仍旧会判断线程池的状态，如果已经不是running亦或是shutdwon，则会直接确保线程被中断。如果没有，将会继续执行并确保不被中断。

接下来可见，我们所需要的任务，直接在工作线程中直接以run()方式以非线程的方式所调用，这里也就是我们所需要的任务真正执行的地方。

在执行完毕后，工作线程的使命并没有真正宣告段落。在while部分worker仍旧会通过getTask()方法试图取得新的任务。

下面是getTask()的实现

private Runnable getTask() {
    boolean timedOut = false; 
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
 
               if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
 
        boolean timed;            
        for (;;) {
            int wc = workerCountOf(c);
            timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
            if (wc <= maximumPoolSize && ! (timedOut && timed))
                break;
            if (compareAndDecrementWorkerCount(c))
                return null;
            c = ctl.get();  
            if (runStateOf(c) != rs)
                continue retry;
        }
 
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

Copy after login

首先仍旧会判断线程池的状态是否是running还是shutdown以及stop状态下队列是否仍旧有需要等待执行的任务。如果状态没有问题，则会跟据allowCoreThreadTimeOut和corePoolSize的值通过对前面这两个属性解释的方式来选择从任务队列中获得任务的方式（是否设置timeout）。其中的timedOut保证了确认前一次试图取任务时超时发生的记录，以确保工作线程的回收。

在runWorker()方法的最后

调用了processWorkerExist()方法来执行工作线程的回收。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) 
        decrementWorkerCount();
 
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
 
    tryTerminate(); 
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; 
        }
        addWorker(null, false);
    }
}

Copy after login

在这一方法中，首先确保已经重新更新了线程池中工作线程的数量，之后从线程池中的工作线程容器移去当前工作线程，并且将完成的任务总数加到线程池的任务总数当中。

在最后仍旧要确保线程池中依旧存在大于等于最小线程数量的工作线程数量存在，如果没有，则重新建立工作线程去等待处理任务队列中任务。

The above is the detailed content of How to use Java thread pool execute() method. For more information, please follow other related articles on the PHP Chinese website!