Java线程池---基本运用到源码解析

序言

近日后台需要一些数据，需要从网上爬取，但是爬取的过程中，由于访问速度太频繁，造成IP被封，最终通过线程池解决；想要借此机会，总结一下线程池的有关知识

线程池框架

ThreadPool_UML.png

从图中可以看出，Executor提供了整个线程池的基本实现，而Executors则提供了各线程池方法的基本工厂方法；下面我们将从线程池的基本运用到源码解析逐一分析

Executor

一个接口，也是线程池框架的基础；方法只有一个execute(),可以用于自定义继承实现任务的执行方式；直接已知子类有：ExecutorService, ScheduledExecutorService
去耦合，可以将任务(Runnable实现)与执行分离
不严格要求异步，可以立即执行，看自己怎么实现；如文档demo：

  class DirectExecutor implements Executor {
    public void execute(Runnable r) {
     r.run(); //此时该任务是在调用的线程中执行的，并非异步
   }
 }

更多的还是要求异步执行

  class ThreadPerTaskExecutor implements Executor {
     public void execute(Runnable r) {
       new Thread(r).start(); //新开辟条线程执行任务
     }
   }

ExecutorService

一个接口，继承了Executor，但是与Executor不同的是，它还提供了终止的方法：shutdown()可以使对象拒绝再接受新的任务，但是以前提交的任务还是会继续执行；shutdownNow()不仅可以使对象拒绝接受新的任务，还尝试阻止以前已经提交的任务开始执行，同时该方法还会试图中断当前正在执行的任务
一个不再使用的ExecutorService应该被关闭，从而释放它所占有的资源
submit()方法扩展了原始的Executor的execute(Runnable)方法,可以返回一个Future对象，该Future对象代表了该等待执行的任务，可以通过该对象来取消执行或者判断是否成功执行(Future的get()方法可以判断)

ScheduledExecutorService

继承自ExecutorService的一个接口，实现类为ScheduledThreadPoolExecutor
可以使任务定期执行或者延迟一段时间之后执行
schedule()方法可以创建各种延时的任务，同时返回一个代表该任务的对象()ScheduledFuture用于检测任务执行情况和取消执行
scheduleWithFixedDelay()方法可以创建和执行任务，该任务具有定时性
当任务通过Executor.execute(Runnable)或者ExecutorService的submit()方法提交时，默认的延时是0；零延时和负延时都被当做立即执行对待
所有的定时方法都支持相对延时或者时间段作为参数，而不是绝对时间或者日期，示例：

schedule(task, date.getTime() - System.currentTimeMillis(), TimeUnit.MILLISECONDS)

官方示例代码：即创建一个任务每10s执行一次，执行1h后取消

import static java.util.concurrent.TimeUnit.*;
class BeeperControl {
    private final ScheduledExecutorService scheduler =
      Executors.newScheduledThreadPool(1);

    public void beepForAnHour() {
       Runnable beeper = () -> System.out.println("beep");
        ScheduledFuture<?> beeperHandle =
        scheduler.scheduleAtFixedRate(beeper, 10, 10, SECONDS);
      Runnable canceller = () -> beeperHandle.cancel(true);
      scheduler.schedule(canceller, 1, HOURS);
    }
}

Future

一个泛型接口，用于表示一个异步执行的结果，可以通过该对象来判断异步是否执行完成(isDone())；同时也可以通过该对象来取回异步执行的返回结果(只能通过get()方法取回，需要判断异步是否完成，get()方法将会阻塞直到任务完成)；也可以通过cancel()方法来取消该任务的执行

Callable<V>

泛型接口，V表示最终异步执行结果的返回类型
与Runnable的区别是：Callable可以返回结果，同时Callable还可以抛出一个可检查的异常(当无法正常返回结果时)，但是Runnable不行

Executors

直接继承自Object，拥有许多工厂方法以支持Executor, ExecutorService,ScheduledExecutorService,ThreadFactory,和Callable对象

对 ExecutorService 的支持

newCachedThreadPool()

使用newCachedThreadPool()的静态方法返回一个新创建的线程池对象；此方法会创建一个线程池，同时该线程池中的线程会被重用(如果空闲可用的话)，如果没有可用的线程，会新建一个线程并添加到该线程池中;该方法适用于需要执行许多耗时短暂的任务集(使线程得到最大程度上的复用，提高程序性能)
注意该线程池中的线程如果在60s内没有被重用的话，将被回收，同时移除出该线程池；所以不用担心因为线程池空闲而造成对系统资源的消耗
实际上是返回一个ThreadPoolExecutor对象，而ThreadPoolExecutor继承于ExecutorService(源代码见下)；实际使用的构造函数是public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue)，表示的是创建一个无界线程池(即线程池中线程数量没有固定)；注意设置的60L对应的是long keepAliveTime参数，表示的是如果线程池中空闲线程数量大于corePoolSize的话，在keepAliveTime时间内没有该线程没有重用，那么会自动回收该线程，同时这里将corePoolSize设置为0，那么在任务执行完成之后所有线程将会自动回收；至于BlockingQueue<Runnable> workQueue)参数，只是一个用于维持由execute()方法提交的任务队列

   public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

示例代码如下；下列代码的输出如下；根据输出可以看出，其实线程池中并没有创建10条线程，正真的最大活跃线程数量只有7条，同时也并不是按照我们预期的那样，任务的执行是分配到特定线程的，而是当thread-1执行完其本身的任务后，线程池中有空闲线程(thread-1)，所以此时并没有使用其他线程去执行任务，而是复用了thread-1。

//示例输出
index...0--Thread info Thread[pool-1-thread-1,5,main]--Thread activeCount 4
index...2--Thread info Thread[pool-1-thread-1,5,main]--Thread activeCount 5
index...1--Thread info Thread[pool-1-thread-2,5,main]--Thread activeCount 5
index...5--Thread info Thread[pool-1-thread-1,5,main]--Thread activeCount 6
index...4--Thread info Thread[pool-1-thread-2,5,main]--Thread activeCount 6
index...8--Thread info Thread[pool-1-thread-1,5,main]--Thread activeCount 7
index...6--Thread info Thread[pool-1-thread-4,5,main]--Thread activeCount 7
index...9--Thread info Thread[pool-1-thread-5,5,main]--Thread activeCount 7
index...3--Thread info Thread[pool-1-thread-3,5,main]--Thread activeCount 7
index...7--Thread info Thread[pool-1-thread-2,5,main]--Thread activeCount 7

  //示例代码，同时注意调用`shutdownNow()`方法后
      public class demo {
          public static void main(String[] args) throws   ExecutionException,         InterruptedException {
              ExecutorService service  =Executors.newCachedThreadPool();
              for(int i=0;i<10;i++){
                  service.execute(new CallDemo(i));
              }
          }
      }

  public class CallDemo implements Runnable {
      private final int index;

      public CallDemo(int index) {
          this.index = index;
      }

      @Override
      public void run() {
          System.out.println("index..."+index+
                "--Thread info "+ Thread.currentThread()+"--Thread         activeCount "+Thread.activeCount());


      }
  }

(如上代码)另外还需要注意的是，当所有任务都执行完了以后，线程池并不是马上停止的，而是此时所有的线程处于空闲状态，会如上所说的等待60s之后再自动回收停止(如果想要任务执行完成之后马上停止，可以调用shutdownNow()或者shutdown()方法)
另外说一下Thread[pool-1-thread-1,5,main]代表的意思(由Thread.currentThread()输出，实际上currentThread()方法返回的是一个Thread对象，只不过在println()函数中会默认调用toString()方法，所以应该看Thread的toString()方法，如下)，对应下面的代码应该可以很清楚各位置输出代表的意思了
```
public String toString() {
      ThreadGroup group = getThreadGroup();
      if (group != null) {
          return "Thread[" + getName() + "," + getPriority() + "," +
                         group.getName() + "]";
      } else {
          return "Thread[" + getName() + "," + getPriority() + "," +
                          "" + "]";
      }
  }
```

至于在代码中使用service.execute(new CallDemo(i));与使用service.submit(new CallDemo(i));的区别，需要注意的是ThreadPoolExecutor类并没有重写submit()方法，所以应该在其父类AbstractExecutorService中找，如下：
可见，其实submit()中也是调用了execute()方法，只是还可以通过Future来取得返回值或者检测任务执行情况(具体见上Future说明)

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
   //null这里可以看出当没有返回值时会默认返回null
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

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

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

newCachedThreadPool(ThreadFactory threadFactory)

参数ThreadFactory对象用于创建新线程；该方法可以根据需要创建线程，所谓的根据需要是指可以通过ThreadFactory创建所需要的线程；至于ThreadFactory可以使程序能够使用其他的线程子类和属性等，而不仅局限与new Thread()，示例：
```
 class SimpleThreadFactory implements ThreadFactory {
  public Thread newThread(Runnable r) {
      //在这里也可以对该线程做一些共同的初始化工作
      return new Thread(r);
    }
  }
```
该线程池中的线程也会被复用
适用于对线程有特殊初始条件需求或者特殊属性需求时

newFixedThreadPool(int nThreads)

会创建一个具有nThread条的线程池，如果待处理事件多余nThread，同时工作的线程任然只有nThread条，剩余的事件将会等待线程空闲
如果线程池中的线程因为事件异常而终止，同时还有待处理事件，那么会重新创建一条线程(也就是说保持线程池中的线程数量为nThread)
该线程池中线程需要明确的调用shutdown()来终止
还有一个newFixedThreadPool(int nThreads, ThreadFactory threadFactory)，其中ThreadFactory的使用同newCachedThreadPool
这个返回的任然是一个ThreadPoolExecutor对象，调用构造函数为ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)，这里将corePoolSize设置为了传入的线程数量，keepAliveTime参数相当于没有起到限定作用(需要注意的是keepAliveTime参数只有在线程池中线程数量大于corePoolSize的时候才会起限定作用，但是这里不存在线程数量超过corePoolSize的情况，所以相当于没有用)，所以使用该方法的时候需要我们显示的调用shutdown()和shutdownNow()方法来回收系统资源(但是需要注意的是如果调用shutdownNow()方法的话，有可能会导致任务没有执行完，因为shutdownNow()会试图终止正在执行的线程，但是shutdown()是会等待当前任务完成的)
```
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
 }
```

newSingleThreadExecutor()

线程池中只有一条线程，如果该线程因为事件异常而终止，而事件队列中还有待处理事件，那么会重新创建一条线程
可以保证事件顺序按照提交顺序执行
同样可以根据需要，使用newSingleThreadExecutor(ThreadFactory threadFactory)来创建特定的线程

至于其与newFixedThreadPool(1)的区别，参见下面代码

ExecutorService executors = Executors.newFixedThreadPool(1);
System.out.println(((ThreadPoolExecutor)     executors).getCorePoolSize());//1
ThreadPoolExecutor executor = (ThreadPoolExecutor) executors;
executor.setCorePoolSize(4);
System.out.println(executor.getCorePoolSize() + "..." +     ((ThreadPoolExecutor) executors).getCorePoolSize());//4...4
executors.shutdownNow();

通过上面代码，可见，newFixedThreadPool(1)线程池中线程的数量是可以调整的，但是对于newSingleThreadExecutor()线程池中线程的数量是不可调整的，如下：

  ExecutorService single = Executors.newSingleThreadExecutor();
  ThreadPoolExecutor singles = (ThreadPoolExecutor) single;
  singles.setCorePoolSize(4);
  System.out.println(((ThreadPoolExecutor) single).getCorePoolSize());

这样将会报类型转换错误：

  java.util.concurrent.Executors$FinalizableDelegatedExecutorService     cannot be cast to java.util.concurrent.ThreadPoolExecutor

实际上该函数返回的是一个FinalizableDelegatedExecutorService对象，源码见下；其中FinalizableDelegatedExecutorService是Executors类中的一个静态包装类，使用了默认访问权限，只能同包中类才能访问

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

这里虽然使用了构造函数ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)，但是实际上是不需要显示调用shutdown()来显示回收系统资源的，因为在包装类FinalizableDelegatedExecutorService中，重写了Object类的finalize()方法，在其中已经调用了shutdown()，而finalize()方法是在垃圾回收时系统自动调用的。其实现如下(但是经过实验，垃圾回收这种情况是不定的(可能涉及到垃圾回收的相关知识)，所以笔者建议还是在不需要时显示调用shutdown()要好一些)
```
static class FinalizableDelegatedExecutorService extends     DelegatedExecutorService {
    FinalizableDelegatedExecutorService(ExecutorService executor) {
        super(executor);
    }
    protected void finalize() {
        super.shutdown(); //这里已经自动调用了shutdown，在垃圾回收的时        候会自动释放系统资源   
    }
}
```

newWorkStealingPool() 与 newWorkStealingPool(int)

核心为work-stealing,意为工作窃取，意为当一个线程完成自己的任务后，去其他的线程队列中窃取任务执行（即将一个大任务分解为数个小任务，为了减小线程执行任务时造成的竞争(同步锁等)，将这些小任务放在不同的双端队列中，当一个线程执行完任务后去其他队列末端(这也是设计成双端队列的原因)取任务来执行），任务结果放在另一个独立队列中。
使用线程池作为执行任务线程的管理者，来实现fork/join模式的任务执行策略，所谓[图片上传失败...(image-d9e8ea-1517904101171)]，就是一种利用一台计算机上的多个处理器进行同类型问题并行计算的模式，通过对大规模问题逐步分解，直到可以作为独立的小任务在单独的线程中执行，结合线程间的通信机制实现相当于递归迭代的并行版本，这和现在流行的Map/Reduce模式有些类似，只不过Map/Reduce是在多台计算机上执行的。---[图片上传失败...(image-52f120-1517904101172)]
可通过Runtime类的availableProcessors方法得到虚拟机的可用处理器数量
```
Runtime run = Runtime.getRuntime();
System.out.println(run.availableProcessors());
```
无法保证执行顺序

unconfigurableExecutorService(ExecutorService executor)

主要用于包装线程池，包装后的线程池不能被修改(属性等)，相当于final,实际上最终代码里面是将传入的executor赋值给了一个private final ExecutorService e成员变量，从而保证线程池属性等不可变

对 ThreadFactory 的支持

ThreadFactory顾名思义是一个用于创建线程的线程工厂方法，可以根据需要创建线程

defaultThreadFactory()

使用默认线程工厂创建线程，这些线程都属于同一线程组，并且都是非守护线程

对 ScheduledExecutorService 的支持

newScheduledThreadPool(int corePoolSize)和

newScheduledThreadPool(int corePoolSize,ThreadFactory threadFactory)

创建指定数量线程的线程池，能够延时或者定期执行任务

本质上该方法返回的是一个ScheduledThreadPoolExecutor对象，而ScheduledThreadPoolExecutor对象调用的又是其父类ThreadPoolExecutor的构造方法(如下)，通过其传入的参数可知，实际上我们最终得到的是一个具有corePoolSize大小的ThreadPoolExecutor线程池

 //构造函数
public static ScheduledExecutorService     newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

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

这里ScheduledThreadPoolExecutor在ThreadPoolExecutor的增加了定时执行任务的功能，即核心scheduleWithFixedDelay(),scheduleAtFixedRate()方法，同样可以通过其返回的ScheduledFuture对象来对任务进行判断是否完成和尝试取消任务,简单使用如下(CallDemo为自定义实现了Runnable的类)：
```
  ScheduledExecutorService service =     Executors.newScheduledThreadPool(5);
  service.scheduleWithFixedDelay(new       CallDemo(1),5000l,1000l,TimeUnit.MILLISECONDS);
  service.scheduleAtFixedRate(new     CallDemo(2),5000l,1000l,TimeUnit.MILLISECONDS);
```
需要注意的时该任务会一直执行下去，除非发生一下情况：1. 明确的通过其返回值ScheduledFuture来取消或者终止；2. 执行的任务产生异常；其中对于第二种情况，使用ScheduledFuture的isDone()会得到true，而且使用get()会跑出ExecutionException异常
至于scheduleWithFixedDelay()和scheduleAtFixedRate()的区别，前者的延时是对前一段任务结束开始计算，后者的延时是前一段任务开始就开始计算了，特别时当任务比较耗时，超过了设定的延时时，前者任然会保持该延时执行，但是后者会等待前一段任务完成之后立即执行，但是不会同时执行

newSingleThreadScheduledExecutor() 和

newSingleThreadScheduledExecutor(ThreadFactory threadFactory)

创建一个单一线程的线程池延时或者定期执行任务
同样当该线程因为事件异常而终止时，在shutdown()之前会再创建一个线程来代替
与newScheduledThreadPool(1)的区别同样也是通过newSingleThreadScheduledExecutor()创建出来的线程池不允许增加新线程和其他属性

unconfigurableScheduledExecutorService(ScheduledExecutorService executor)

同样用于包装线程池，相当与final

对 Callable 的支持

Callable<Object> callable(Runnable task)

执行任务，同时返回null

<T> Callable<T> callable(Runnable task,T result)

执行指定返回结果result，这在需要调用Callable但是结果无返回值时的会很有用

Callable<Object> callable(PrivilegedAction<?> action)

参考博客

最后编辑于：2018.02.06 16:13:50

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,602评论 6赞 481
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,442评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 152,878评论 0赞 344
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,306评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,330评论 5赞 373
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,071评论 1赞 285
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,382评论 3赞 400
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,006评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,512评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,965评论 2赞 325
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,094评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,732评论 4赞 323
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,283评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,286评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,512评论 1赞 262
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,536评论 2赞 354
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,828评论 2赞 345