java - newCachedThreadPool 是如何复用线程的？

Question

我想就 Zeller 一年多前发布的同一个问题询问更多细节...

javadoc 说 Executors.newCachedThreadPool 返回的服务重用了线程。这怎么可能？

我知道队列结构是如何在内部设置的，但我没有看到它如何重用队列中的线程。

我见过的所有示例都是让开发人员创建他们的线程实例并通过“执行”方法将其传递。

例如……

ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {            
  Runnable worker = new WorkerThread(i);  //will create 10 instances          
  executor.execute(worker);          
} 

我知道线程池可以轻松管理每个线程的生命周期，但同样，我看不到任何方法，也看不到访问或重启池中任何线程的能力。

在上面的示例中，我希望每个线程都将由线程池启动、运行、终止和处理，但绝不会重用。

消息传递系统就是您需要它的示例。假设您有一个 onMessage 处理程序，并且您希望重用池中的线程之一来处理它，所以我期望像...这样的方法

worker = executor.getIdleThread;
worker.setData(message);
executor.resubmit(worker);          

或者让 ExecutorService 充当工厂类并让它返回您的线程的一个实例，它在内部决定创建一个新线程或重用旧线程。

ExecutorService executor = Executors.newCachedThreadPool(WorkerThread);
Runnable worker = executor.getThread;
worker.setData(message);

所以我错过了一些东西。这可能很简单，但我花了一个下午阅读教程和示例，但仍然没有弄明白。有人可以阐明这个主题吗？

Answer 1

我也很好奇这怎么可能，因为Threads无法重启，所以我分析了ThreadPoolExecutor的代码，它是所有ThreadPool ExecutorService的实现通过静态构造函数。

首先，如其他答案所述，您不使用线程，而是使用线程池中的可运行对象，因为那样会破坏目的。所以这里详细解释了一个 ExecutorService 是如何重用线程的:

您通常通过 submit() 添加一个 Runnable，它在内部调用 execute() 方法。基本上这会将 runnable 添加到队列中，如果没有工作的 ATM，则添加一个 Worker

public void execute(Runnable command) {
        ...
        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))
            reje
ct(command);
    }

执行器维护了一堆Worker(ThreadPoolExecutor的内部类)。它具有您提交的可运行对象和一个线程，该线程将通过您可能设置的 ThreadFactory 创建，或者只是一个默认线程； Worker本身也是一个Runnable，用于从工厂创建Thread

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
...
   Worker(Runnable firstTask) {
     this.firstTask = firstTask;
     this.thread = getThreadFactory().newThread(this);
   }

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

添加 Worker 时它会启动

private boolean addWorker(Runnable firstTask, boolean core) {
    ...
    Worker w = new Worker(firstTask);
    Thread t = w.thread;
    ...
    t.start();
    ...
    return true;
}

runWorker() 方法在循环中运行并通过 getTask() 获取您提交的在 workingQueue 中排队的可运行对象，并且将在 getTask 处等待，直到发生超时。

   final void runWorker(Worker w) {
        Runnable task = w.firstTask;
        w.firstTask = null;
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                clearInterruptsForTaskRun();
                try {
                    beforeExecute(w.thread, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                    ...
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

这里是 getTask() 方法

private Runnable getTask() {
   ...
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            ..
        } catch (InterruptedException retry) {
            ...
        }
    }
}

---

tl;dr 所以基本上，Threadpool 维护工作线程，这些工作线程在循环中运行并执行由阻塞队列提供的可运行对象。 worker 将根据需求创建和销毁(没有更多任务，worker 将结束；如果没有空闲 worker 并且 < maxPoolSize 则创建新的 worker)。此外，我不会将其称为“重用”，因为线程将用作循环程序来执行所有可运行对象。