使用Executors创建线程池

Executor是一个工厂类，可以直接创建线程池，从最简单的定长线程池开始学习

public static ExecutorService newFixedThreadPool(int nThreads) {

return new ThreadPoolExecutor(nThreads, nThreads,

0L, TimeUnit.MILLISECONDS,

new LinkedBlockingQueue());

}

复制代码

可以看到真正创建线程池的方法是调用ThreadPollExecutor的构造方法

public ThreadPoolExecutor(int corePoolSize, //核心线程数

int maximumPoolSize, //最大线程数

long keepAliveTime, //非核心线程的空闲存货时间

TimeUnit unit, //时间单位

BlockingQueue workQueue //等待队列

) {

this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,

Executors.defaultThreadFactory() //创建线程池的工厂, defaultHandler //拒绝策略);

}

复制代码

当创建一个线程池后就可以，想线程池中提交任务了,先看ThreadPollExecutor顶层接口的java.util.concurrent.Executor#execute方法

public void execute(Runnable command) {

if (command == null)

throw new NullPointerException();

/*

* Proceed in 3 steps:

*

* 1. If fewer than corePoolSize threads are running, try to

* start a new thread with the given command as its first

* task. The call to addWorker atomically checks runState and

* workerCount, and so prevents false alarms that would add

* threads when it shouldn't, by returning false.

* 如过还有空闲的核心线程，尝试开启一个新线程执行任务，addWorker时还会检测runState和线程数，防止添加不应该添加的线程。

*

* 2. If a task can be successfully queued, then we still need

* to double-check whether we should have added a thread

* (because existing ones died since last checking) or that

* the pool shut down since entry into this method. So we

* recheck state and if necessary roll back the enqueuing if

* stopped, or start a new thread if there are none.

*

*

* 3. If we cannot queue task, then we try to add a new

* thread. If it fails, we know we are shut down or saturated

* and so reject the task.

* 入队失败

*/

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

}

复制代码

接下来看下java.util.concurrent.ThreadPoolExecutor#addWorker方法

private boolean addWorker(Runnable firstTask, boolean core) {

retry:

for (;;) {

int c = ctl.get();

int rs = runStateOf(c);

// Check if queue empty only if necessary.

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)) //cas修改运行线程数

break retry;

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

if (runStateOf(c) != rs) //检查运行状态

continue retry;

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

mainLock.lock();

try {

// Recheck while holding lock.

// Back out on ThreadFactory failure or if

// shut down before lock acquired.

int rs = runStateOf(ctl.get());

if (rs < SHUTDOWN ||

(rs == SHUTDOWN && firstTask == null)) {

if (t.isAlive()) // precheck that t is startable

throw new IllegalThreadStateException();

workers.add(w); //添加的Set集合

int s = workers.size();

if (s > largestPoolSize)

largestPoolSize = s;

workerAdded = true;

}

} finally {

mainLock.unlock();

}

if (workerAdded) {

t.start(); //调用start方法

workerStarted = true;

}

}

} finally {

if (! workerStarted)

addWorkerFailed(w);

}

return workerStarted;

}

复制代码

worker的定义

private final class Worker

extends AbstractQueuedSynchronizer

implements Runnable

复制代码

execute没有返回值，如果想获取返回值，可以使用java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable)

public Future> submit(Runnable task) {

if (task == null) throw new NullPointerException();

RunnableFuture ftask = newTaskFor(task, null);

execute(ftask);

return ftask;

}

复制代码

执行线程池中的所有任务可以使用java.util.concurrent.AbstractExecutorService#invokeAll(java.util.Collection extends java.util.concurrent.Callable>)

public List> invokeAll(Collection extends Callable> tasks)

throws InterruptedException {

if (tasks == null)

throw new NullPointerException();

ArrayList> futures = new ArrayList>(tasks.size());

boolean done = false;

try {

for (Callable t : tasks) { //遍历tasks 执行execute()

RunnableFuture f = newTaskFor(t);

futures.add(f);

execute(f);

}

for (int i = 0, size = futures.size(); i < size; i++) {

Future f = futures.get(i);

if (!f.isDone()) {

try {

f.get();

} catch (CancellationException ignore) {

} catch (ExecutionException ignore) {

}

}

}

done = true;

return futures;

} finally {

if (!done)

for (int i = 0, size = futures.size(); i < size; i++)

futures.get(i).cancel(true);

}

}

复制代码

那么线程池在哪里实现了复用呢？上面可以看到addWorker方法中执行的是Worker的run()方法，真正执行的是java.util.concurrent.ThreadPoolExecutor#runWorker，然后循环执行任务

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

}

复制代码

在java.util.concurrent.ThreadPoolExecutor#getTask中，当活动线程数不大于最大线程数时，就会从队列中获取任务，如果获取不到就阻塞，这样核心线程就不会被回收，每当任务来时也能直接执行，不用再创建线程

boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

//..

try {

Runnable r = timed ?

workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :

workQueue.take();

if (r != null)

return r;

timedOut = true;

} catch (InterruptedException retry) {

timedOut = false;

}

复制代码

待续。。