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

作者: HusterYP | 来源:发表于2018-02-06 16:01 被阅读0次

序言

近日后台需要一些数据，需要从网上爬取，但是爬取的过程中，由于访问速度太频繁，造成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但是结果无返回值时的会很有用

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

序言

线程池框架

Executor

ExecutorService

ScheduledExecutorService

Future

Callable<V>

Executors

对 ExecutorService 的支持

newCachedThreadPool()

newCachedThreadPool(ThreadFactory threadFactory)

newFixedThreadPool(int nThreads)

newSingleThreadExecutor()

newWorkStealingPool() 与 newWorkStealingPool(int)

unconfigurableExecutorService(ExecutorService executor)

对 ThreadFactory 的支持

defaultThreadFactory()

对 ScheduledExecutorService 的支持

newScheduledThreadPool(int corePoolSize)和

newSingleThreadScheduledExecutor() 和

unconfigurableScheduledExecutorService(ScheduledExecutorService executor)

对 Callable 的支持

Callable<Object> callable(Runnable task)

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

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

参考博客

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 杂谈

程序员

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

序言

线程池框架

Executor

ExecutorService

ScheduledExecutorService

Future

Callable<V>

Executors

对 ExecutorService 的支持

newCachedThreadPool()

newCachedThreadPool(ThreadFactory threadFactory)

newFixedThreadPool​(int nThreads)

newSingleThreadExecutor​()

newWorkStealingPool​() 与 newWorkStealingPool(int)

unconfigurableExecutorService(ExecutorService executor)

对 ThreadFactory 的支持

defaultThreadFactory()

对 ScheduledExecutorService 的支持

newScheduledThreadPool​(int corePoolSize)和

newSingleThreadScheduledExecutor​() 和

unconfigurableScheduledExecutorService​(ScheduledExecutorService executor)

对 Callable 的支持

Callable<Object> callable​(Runnable task)

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

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

参考博客

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 杂谈

程序员

newFixedThreadPool(int nThreads)

newSingleThreadExecutor()

newWorkStealingPool() 与 newWorkStealingPool(int)

newScheduledThreadPool(int corePoolSize)和

newSingleThreadScheduledExecutor() 和

unconfigurableScheduledExecutorService(ScheduledExecutorService executor)

Callable<Object> callable(Runnable task)

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

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