Concurrent包中的线程池

大家好，欢迎来到IT知识分享网。

线程的生成

在corrent包中为我们提供了线程池的实现类Executors。Executors生成线程池有以下几种方式：

//固定线程数线程池 1.newFixedThreadPool(int nThreads) 2.newFixedThreadPool(int nThreads, ThreadFactory threadFactory) //一个线程的线程池 3.newSingleThreadExecutor() 4.newSingleThreadExecutor(ThreadFactory threadFactory) //0~无限多个线程的线程池 5.newCachedThreadPool() 6.newCachedThreadPool(ThreadFactory threadFactory) //一个定时执行任务的线程池 7newSingleThreadScheduledExecutor(ThreadFactory threadFactory) //指定数目定时执行任务的线程池 8.newScheduledThreadPool(int corePoolSize) 9.ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)

方法中的1-6的具体实现方法都相同，只是参数不同而已。接下来看一下具体实现方法：

public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads,nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }

通过上述代码可以看出，线程池的具体的实现类为ThreadPoolExecutor，任务存储结构为LinkedBlockingQueue。

线程的执行

线程池的生成方法都返回一个ExecutorService类，这个类提供线程进入入口：
Future<?> submit(Runnable task)
接下来看submit的执行方法：

 public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; }

上面返回的结果为Future类为任务类FutureTask返回的结果，通过future.get()方法获取返回结果。接下来看具体的执行方法ThreadPoolExecutor.execute(Runnable command)：

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

首先通过方法参数可以知道，执行任务需要封装成Runable对象，继而我们可以知道生成的线程的方法是new Thread(Runnable runnable);之后我们看一下执行的逻辑，执行线程分为三步：
1.获取运行的线程数，如果工作的线程数小于核心线程数，尝试添加线程到线程池中
2.添加线程后，执行double-check,查看线程池是否关闭，线程是否添加到池中
3.如果2执行失败，再次尝试添加线程到线程池中，如果失败，拒绝任务
接下来看看添加线程的详细的代码

 private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { //获取运行线程数，ctl为AtomicInteger保证并发执行的安全性 int c = ctl.get(); int rs = runStateOf(c); //检测线程任务对列是否为空，如果任务对列为空，直接返回false 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(); 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 { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int c = ctl.get(); int rs = runStateOf(c); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //workers为HashSet类型 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; }

在添加线程到线程池的这个过程中，获取运行线程数多次，可见添加运行线程复杂。

线程的回收

最后来看一下线程池的回收：

 protected void finalize() { shutdown(); } public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN); interruptIdleWorkers(); onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } //如果有运行的线程池，则直接返回，否则告诉所有线程状态为停止态 tryTerminate(); } 

整个线程池的由初始化到执行过程，再到结束的过程就总结到这里了。