一.线程保活和销毁

创建线程

  1. Worker w = null;
  2. w = new Worker(firstTask);
  3. final Thread t = w.thread;
  4. // 执行Worker的run方法
  5. t.start();
  7. Worker(Runnable firstTask) {
  8.     this.firstTask = firstTask;
  9.     this.thread = getThreadFactory().newThread(this);
  10. }
  12. public Thread newThread(Runnable r) {
  13.     Thread t = new Thread(group, r,
  14.                           namePrefix + threadNumber.getAndIncrement(),
  15.                           0);
  16.     return t;
  17. }

Worker的run方法

  1. public void run() {
  2.     runWorker(this);
  3. }
  5. final void runWorker(Worker w) {
  6.     // firstTask是execute方法的入参
  7.     Runnable task = w.firstTask;
  8.     w.firstTask = null;
  9.     while (task != null || (task = getTask()) != null) {
  10.         try {
  11.             task.run();
  12.         } finally {
  13.             task = null;
  14.         }
  15.     } 
  16. }
  18. private Runnable getTask() {
  19.     boolean timedOut = false;
  20.     for (;;) {
  21.         int c = ctl.get();
  22.         // 线程数
  23.         int wc = workerCountOf(c);
  24.         // allowCoreThreadTimeOut设置为true，核心线程可销毁
  25.         // wc > corePoolSize，超出核心线程数的部分
  26.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  27.         // 长时间空闲返回null，线程销毁
  28.         if (timed && timedOut) {
  29.             if (compareAndDecrementWorkerCount(c))
  30.                 return null;
  31.             continue;
  32.         }
  33.         try {
  34.             Runnable r = timed ?
  35.                 // 没有数据返回null
  36.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  37.                 // 阻塞直到有数据
  38.                 workQueue.take();
  39.             if (r != null)
  40.                 return r;
  41.             timedOut = true;
  42.         } catch (InterruptedException retry) {
  43.             timedOut = false;
  44.         }
  45.     }
  46. }

二.DiscardOldestPolicy

丢弃队列最前面的任务，然后重新提交被拒绝的任务。

  1. final void reject(Runnable command) {
  2.     handler.rejectedExecution(command, this);
  3. }
  6. public static class DiscardOldestPolicy implements RejectedExecutionHandler {
  7.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  8.         if (!e.isShutdown()) {
  9.             // 删除队首元素
  10.             e.getQueue().poll();
  11.             e.execute(r);
  12.         }
  13.     }
  14. }
  16. public void execute(Runnable command) {
  17.     int c = ctl.get();
  18.     if (workerCountOf(c) < corePoolSize) {
  19.         if (addWorker(command, true))
  20.             return;
  21.         c = ctl.get();
  22.     }
  23.     // 可能执行到这，队首元素被删了，可以入队了
  24.     if (isRunning(c) && workQueue.offer(command)) {
  26.     }
  27.     else if (!addWorker(command, false))
  28.         reject(command);
  29. }

�三.自带线程池

1.newSingleThreadExecutor
单线程，所有任务的执行顺序按照任务的提交顺序执行。

  1. public class Test {
  3.     public static void main(String[] args) {
  4.         ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
  5.         for (int i = 0; i < 10; i++) {
  6.             final int index = i;
  7.             singleThreadExecutor.execute(new Runnable() {
  8.                 @SneakyThrows
  9.                 @Override
  10.                 public void run() {
  11.                     System.out.println(index);
  12.                     Thread.sleep(2000);
  13.                 }
  14.             });
  15.         }
  16.     }
  17. }
  19. 输出：
  20. 0
  21. 1
  22. 2
  23. 3
  24. 4
  25. 5
  26. 6
  27. 7
  28. 8
  29. 9
  1. public static ExecutorService newSingleThreadExecutor() {
  2.     return new FinalizableDelegatedExecutorService
  3.         (new ThreadPoolExecutor(1, 1,
  4.                                 0L, TimeUnit.MILLISECONDS,
  5.                                 new LinkedBlockingQueue<Runnable>()));
  6. }

2.newFixedThreadPool
定长线程池

  1. public class Test {
  3.     public static void main(String[] args) {
  4.         ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
  5.         for (int i = 0; i < 10; i++) {
  6.             final int index = i;
  7.             fixedThreadPool.execute(new Runnable() {
  9.                 @SneakyThrows
  10.                 @Override
  11.                 public void run() {
  12.                     System.out.println(index + ":" + System.currentTimeMillis());
  13.                     Thread.sleep(3000);
  14.                 }
  15.             });
  16.         }
  17.     }
  18. }
  20. 输出：
  21. 0:1643030738793
  22. 3:1643030738793
  23. 1:1643030738793
  24. 2:1643030738793
  25. 4:1643030738793
  26. 5:1643030741795
  27. 6:1643030741795
  28. 9:1643030741795
  29. 8:1643030741795
  30. 7:1643030741795
  1. public static ExecutorService newFixedThreadPool(int nThreads) {
  2.     return new ThreadPoolExecutor(nThreads, nThreads,
  3.                                   0L, TimeUnit.MILLISECONDS,
  4.                                   new LinkedBlockingQueue<Runnable>());
  5. }

3.newCachedThreadPool
可缓存的线程池，线程可用时重用线程，否则创建新的线程。

  1. public class Test {
  3.     private static AtomicInteger atomicInteger = new AtomicInteger();
  5.     public static void main(String[] args) throws InterruptedException {
  6.         ExecutorService executorService = Executors.newCachedThreadPool();
  7.         for (int i = 0; i < 5; i++) {
  8.             Thread.sleep(100L);
  9.             executorService.submit(Test::print);
  10.         }
  11.     }
  13.     public static void print() {
  14.         System.out.println(Thread.currentThread().getName());
  15.         System.out.println(atomicInteger.incrementAndGet());
  16.     }
  18. }
  20. 输出：
  21. pool-1-thread-1
  22. 1
  23. pool-1-thread-1
  24. 2
  25. pool-1-thread-1
  26. 3
  27. pool-1-thread-1
  28. 4
  29. pool-1-thread-1
  30. 5
  1. public static ExecutorService newCachedThreadPool() {
  2.     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  3.                                   60L, TimeUnit.SECONDS,
  4.                                   new SynchronousQueue<Runnable>());
  5. }
  7. public void execute(Runnable command) {
  8.     int c = ctl.get();
  9.     if (workerCountOf(c) < corePoolSize) {
  10.         if (addWorker(command, true))
  11.             return;
  12.         c = ctl.get();
  13.     }
  14.     // 入队成功，复用线程
  15.     if (isRunning(c) && workQueue.offer(command)) {
  16.         int recheck = ctl.get();
  17.         if (! isRunning(recheck) && remove(command))
  18.             reject(command);
  19.         else if (workerCountOf(recheck) == 0)
  20.             addWorker(null, false);
  21.     }
  22.     // 入队不成功，新建线程
  23.     else if (!addWorker(command, false))
  24.         reject(command);
  25. }
  27. public boolean offer(E e) {
  28.     if (e == null) throw new NullPointerException();
  29.     return transferer.transfer(e, true, 0) != null;
  30. }
  32. E transfer(E e, boolean timed, long nanos) {
  33.     SNode s = null;
  34.     int mode = (e == null) ? REQUEST : DATA;
  35.     for (;;) {
  36.         SNode h = head;
  37.         if (h == null || h.mode == mode) {  
  38.             if (timed && nanos <= 0) {     
  39.                 // 没有线程可用
  40.                 return null;
  41.             }
  42.         } else if (!isFulfilling(h.mode)) { 
  43.             // 有线程可用
  44.             if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
  45.                 for (;;) { 
  46.                     SNode m = s.next;       
  47.                     SNode mn = m.next;
  48.                     if (m.tryMatch(s)) {
  49.                         casHead(s, mn);
  50.                         return (E) ((mode == REQUEST) ? m.item : s.item);
  51.                     }
  52.                 }
  53.             }
  54.         }
  55.     }
  56. }

4.newScheduledThreadPool
支持定时及周期性任务执行。

  1. // 延迟3秒执行
  2. public class Test {
  3.     public static void main(String[] args) throws InterruptedException {
  4.         ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  5.         System.out.println(System.currentTimeMillis());
  6.         scheduledThreadPool.schedule(new Runnable() {
  7.             @Override
  8.             public void run() {
  9.                 System.out.println("********************");
  10.                 System.out.println(System.currentTimeMillis());
  11.             }
  12.         }, 3, TimeUnit.SECONDS);
  13.     }
  14. }
  16. 输出：
  17. 1643101162540
  18. ********************
  19. 1643101165544
  1. public class Test {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  4.         System.out.println(System.currentTimeMillis());
  5.         scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
  6.             @Override
  7.             public void run() {
  8.                 System.out.println("**************");
  9.                 System.out.println(System.currentTimeMillis());
  10.             }
  11.         }, 1, 3, TimeUnit.SECONDS);
  12.     }
  13. }
  15. 输出：
  16. 1643101766597
  17. **************
  18. 1643101767603
  19. **************
  20. 1643101770600
  21. **************
  22. 1643101773602
  23. **************
  24. 1643101776600
  1. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
  2.     return new ScheduledThreadPoolExecutor(corePoolSize);
  3. }
  5. public ScheduledThreadPoolExecutor(int corePoolSize) {
  6.     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  7.           new DelayedWorkQueue());
  8. }
  1. public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
  2.                                               long initialDelay,
  3.                                               long period,
  4.                                               TimeUnit unit) {
  5.     //triggerTime(initialDelay, unit):当前时间+initialDelay，纳秒
  6.     ScheduledFutureTask<Void> sft =
  7.         new ScheduledFutureTask<Void>(command,
  8.                                       null,
  9.                                       triggerTime(initialDelay, unit),
  10.                                       unit.toNanos(period));
  11.     RunnableScheduledFuture<Void> t = decorateTask(command, sft);
  12.     sft.outerTask = t;
  13.     delayedExecute(t);
  14.     return t;
  15. }
  17. ScheduledFutureTask(Runnable r, V result, long ns, long period) {
  18.     super(r, result);
  19.     this.time = ns;
  20.     this.period = period;
  21. }
  23. protected <V> RunnableScheduledFuture<V> decorateTask(
  24.     Runnable runnable, RunnableScheduledFuture<V> task) {
  25.     return task;
  26. }
  28. private void delayedExecute(RunnableScheduledFuture<?> task) {
  29.     super.getQueue().add(task);
  30.     ensurePrestart();
  31. }
  33. void ensurePrestart() {
  34.     int wc = workerCountOf(ctl.get());
  35.     if (wc < corePoolSize)
  36.         addWorker(null, true);
  37.     else if (wc == 0)
  38.         addWorker(null, false);
  39. }
  41. //  private static final int INITIAL_CAPACITY = 16;
  42. //  private RunnableScheduledFuture<?>[] queue =
  43. //      new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
  44. // 小根堆
  45. public boolean offer(Runnable x) {
  46.     if (x == null)
  47.         throw new NullPointerException();
  48.     RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
  49.     final ReentrantLock lock = this.lock;
  50.     lock.lock();
  51.     try {
  52.         int i = size;
  53.         if (i >= queue.length)
  54.             grow();
  55.         size = i + 1;
  56.         if (i == 0) {
  57.             queue[0] = e;
  58.             setIndex(e, 0);
  59.         } else {
  60.             siftUp(i, e);
  61.         }
  62.         if (queue[0] == e) {
  63.             leader = null;
  64.             available.signal();
  65.         }
  66.     } finally {
  67.         lock.unlock();
  68.     }
  69.     return true;
  70. }
  1. public RunnableScheduledFuture<?> take() throws InterruptedException {
  2.     final ReentrantLock lock = this.lock;
  3.     lock.lockInterruptibly();
  4.     try {
  5.         for (;;) {
  6.             RunnableScheduledFuture<?> first = queue[0];
  7.             if (first == null)
  8.                 available.await();
  9.             else {
  10.                 long delay = first.getDelay(NANOSECONDS);
  11.                 if (delay <= 0)
  12.                     return finishPoll(first);
  13.                 first = null;
  14.                 if (leader != null)
  15.                     available.await();
  16.                 else {
  17.                     Thread thisThread = Thread.currentThread();
  18.                     leader = thisThread;
  19.                     try {
  20.                         available.awaitNanos(delay);
  21.                     } finally {
  22.                         if (leader == thisThread)
  23.                             leader = null;
  24.                     }
  25.                 }
  26.             }
  27.         }
  28.     } finally {
  29.         if (leader == null && queue[0] != null)
  30.             available.signal();
  31.         lock.unlock();
  32.     }
  33. }
  35. private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
  36.     int s = --size;
  37.     RunnableScheduledFuture<?> x = queue[s];
  38.     queue[s] = null;
  39.     if (s != 0)
  40.         siftDown(0, x);
  41.     setIndex(f, -1);
  42.     return f;
  43. }

ScheduledFutureTask.run

  1. public void run() {
  2.     // 执行run方法**************
  3.     if (ScheduledFutureTask.super.runAndReset()) {
  4.         setNextRunTime();
  5.         reExecutePeriodic(outerTask);
  6.     }
  7. }
  9. private void setNextRunTime() {
  10.     long p = period;
  11.     if (p > 0)
  12.         time += p;
  13.     else
  14.         time = triggerTime(-p);
  15. }
  17. void reExecutePeriodic(RunnableScheduledFuture<?> task) {
  18.     if (canRunInCurrentRunState(true)) {
  19.         super.getQueue().add(task);
  20.         ensurePrestart();
  21.     }
  22. }