Java 高并发六：JDK并发包2详解

 public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); } public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())); } public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue()); } 

 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }

 import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ThreadPoolDemo { public static class MyTask implements Runnable { @Override public void run() { System.out.println(System.currentTimeMillis() + "Thread ID:" + Thread.currentThread().getId()); try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } } } public static void main(String[] args) { MyTask myTask = new MyTask(); ExecutorService es = Executors.newFixedThreadPool(5); for (int i = 0; i <10; i++) { es.submit(myTask); } } } 

 import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class ThreadPoolDemo { public static void main(String[] args) { ScheduledExecutorService ses = Executors.newScheduledThreadPool(10); //如果前面的任务还未完成，则调度不会启动。 ses.scheduleWithFixedDelay(new Runnable() { @Override public void run() { try { Thread.sleep(1000); System.out.println(System.currentTimeMillis()/1000); } catch (Exception e) { // TODO: handle exception } } }, 0, 2, TimeUnit.SECONDS);//启动0秒后执行，然后周期2秒执行一次 } } 

 ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue()){ @Override protected void beforeExecute(Thread t, Runnable r) { System.out.println("准备执行"); } @Override protected void afterExecute(Runnable r, Throwable t) { System.out.println("执行完成"); } @Override protected void terminated() { System.out.println("线程池退出"); } }; 

 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize <0 || maximumpoolsize < corepoolsize keepalivetime 0) throw new illegalargumentexception(); if (workqueue== null threadfactory== handler== null) nullpointerexception(); this.corepoolsize=corePoolSize; this.maximumpoolsize=maximumPoolSize; this.workqueue=workQueue; this.keepalivetime=unit.toNanos(keepAliveTime); this.threadfactory=threadFactory; this.handler=handler; } pre>

 ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue(), new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { System.out.println(r.toString() + "is discard"); } }); 

 static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } 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; } } 

 import java.util.ArrayList; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinTask; import java.util.concurrent.RecursiveTask; public class CountTask extends RecursiveTask{ private static final int THRESHOLD = 10000; private long start; private long end; public CountTask(long start, long end) { super(); this.start = start; this.end = end; } @Override protected Long compute() { long sum = 0; boolean canCompute = (end - start)  subTasks = new ArrayList(); long pos = start; for (int i = 0; i <100; i++) { long lastOne = pos + step; if(lastOne > end ) { lastOne = end; } CountTask subTask = new CountTask(pos, lastOne); pos += step + 1; subTasks.add(subTask); subTask.fork();//把子任务推向线程池 } for (CountTask t : subTasks) { sum += t.join();//等待所有子任务结束 } } return sum; } public static void main(String[] args) { ForkJoinPool forkJoinPool = new ForkJoinPool(); CountTask task = new CountTask(0, 200000L); ForkJoinTask result = forkJoinPool.submit(task); try { long res = result.get(); System.out.println("sum = " + res); } catch (Exception e) { // TODO: handle exception e.printStackTrace(); } } } 

 final void runWorker(WorkQueue w) { w.growArray(); // allocate queue for (int r = w.hint; scan(w, r) == 0; ) { r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift } }

 private final int scan(WorkQueue w, int r) { WorkQueue[] ws; int m; long c = ctl;    // for consistency check if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 && w != null) { for (int j = m + m + 1, ec = w.eventCount;;) { WorkQueue q; int b, e; ForkJoinTask[] a; ForkJoinTask t; if ((q = ws[(r - j) & m]) != null && (b = q.base) - q.top <0 && (a=q.array) !=null) { long i=(((a.length - 1) & b) << ashift) + abase; if ((t=((ForkJoinTaskNO NUMERIC NOISE KEY 1002)u.getobjectvolatile(a, i))
 ..
 w.runTask(t);
 ..
通过这个大概的分析理解了过程，我们发现，当前线程调用scan方法后，不会执行当前的WorkQueue中的任务，而是通过一个随机数r，来得到其他 WorkQueue的任务。这就是ForkJoinPool的主要的一个机理。
 当前线程不会只着眼于自己的任务，而是优先完成其他任务。这样做来，防止了饥饿现象的发生。这样就预防了某些线程因为卡死或者其他原因而无法及时完成任务，或者某个线程的任务量很大，其他线程却没事可做。
然后来看看runTask方法

 final void runTask(ForkJoinTask task) { if ((currentSteal = task) != null) { ForkJoinWorkerThread thread; task.doExec(); ForkJoinTask[] a = array; int md = mode; ++nsteals; currentSteal = null; if (md != 0) pollAndExecAll(); else if (a != null) { int s, m = a.length - 1; ForkJoinTask t; while ((s = top - 1) - base >= 0 && (t = (ForkJoinTask)U.getAndSetObject (a, ((m & s) << ASHIFT) + ABASE, null)) != null) { top = s; t.doExec(); } } if ((thread = owner) != null) // no need to do in finally clause thread.afterTopLevelExec(); } } 
有一个有趣的命名：currentSteal，偷得的任务，的确是刚刚解释的那样。
task.doExec();
将会完成这个任务。
完成了别人的任务以后，将会完成自己的任务。
通过得到top来获得自己任务第一个任务
 while ((s = top - 1) - base >= 0 && (t = (ForkJoinTask)U.getAndSetObject(a, ((m & s) << ASHIFT) + ABASE, null)) != null) { top = s; t.doExec(); }
接下来，通过一个图来总结下刚刚线程池的流程
比如有T1，T2两个线程，T1会通过T2的base来获得T2的最后一个任务（当然实际上是通过一个随机数r来取得某个线程最后一个任务），T1也会通过自己的top来执行自己的第一个任务。反之，T2也会如此。
 拿其他线程的任务都是从base开始拿的，自己拿自己的任务是从top开始拿的。这样可以减少冲突
如果没有找到其他任务
 else if (--j <0) { if ((ec | (e = (int)c)) <0) // inactive or terminating return awaitWork(w, c, ec); else if (ctl == c) {  // try to inactivate and enqueue long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK)); w.nextWait = e; w.eventCount = ec | INT_SIGN; if (!U.compareAndSwapLong(this, CTL, c, nc)) w.eventCount = ec; // back out } break; } 
那么首先会通过一系列运行来改变ctl的值，获得了nc，然后用CAS将新的值赋值。然后就调用awaitWork()将线程进入等待状态（调用的 前面系列文章中提到的unsafe的park方法）。
 这里要说明的是改变ctl值这里，首先是将ctl中的AC-1，AC是占ctl的前16位，所以不能直接-1，而是通过AC_UNIT（0x1000000000000）来达到使ctl的前16位-1的效果。
前面说过eventCount中有保存poolIndex，通过poolIndex以及WorkQueue中的nextWait，就能遍历所有的等待线程。