线程重用——线程池的基本原理

为简单起见，线程池中只有一个线程：

package com.xs.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ThreadPool {

private final BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<Runnable>();

private final Thread thread;

public ThreadPool() {
thread = ThreadFactory.newThread(blockingQueue);
thread.start();
}

public void execute(Runnable runnable){
try {
blockingQueue.put(runnable);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}

public static void main(String[] args) throws Exception {
ThreadPool pool = new ThreadPool();
Runnable runnable = new Runnable() {

@Override
public void run() {
System.out.println("Hello world!");
}
};
for (int i = 0; i < 10; i++) {
pool.execute(runnable);
Thread.sleep(1000);
}
}
}

ThreadFactory：

package com.xs.concurrent;

import java.util.concurrent.BlockingQueue;

public class ThreadFactory {

public static Thread newThread(final BlockingQueue<Runnable> tasks){
Thread t = new Thread(new Runnable() {

@Override
public void run() {
for (;;) {
try {
Runnable task = tasks.take(); // 阻塞方法，直到取到任务为止
task.run();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
});
return t;
}

private ThreadFactory(){}
}

Thread.start()不能重复调用，所以要重用Thread，就不能让Thread执行完一个任务后终止，因此就必须阻塞Thread.run()方法，让该方法不停地从任务队列中获取任务并执行。

下面看看JDK的源码。

ThreadPoolExecutor.execute()方法：

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}

addIfUnderCorePoolSize()：

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}

这个方法启动了一个线程。

addThread()：

private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w);
if (t != null) {
w.thread = t;
workers.add(w);
int nt = ++poolSize;
if (nt > largestPoolSize)
largestPoolSize = nt;
}
return t;
}

这个方法将任务交给了Worker。Worker是ThreadPoolExecutor的内部类。
Worker.run()：

public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task); // 执行任务
task = null;
}
} finally {
workerDone(this);
}
}

getTask()方法属于ThreadPoolExecutor类：

Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN)  // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers();
return null;
}
// Else retry
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}

该方法不停地从工作队列中获取任务。