程序开发 ·

Java 线程池相关知识点总结

Android中常见到的很多通用组件一般都离不开”池”的概念,如各种图片加载库,网络请求库,即使Android的消息传递机制中的Meaasge当使用Meaasge.obtain()就是使用的Meaasge池中的对象,因此这个概念很重要。本文将介绍的线程池技术同样符合这一思想。

线程池的优点

重用线程池中的线程,减少因对象创建,销毁所带来的性能开销;

能有效的控制线程的最大并发数,提高系统资源利用率,同时避免过多的资源竞争,避免堵塞;

能够多线程进行简单的管理,使线程的使用简单、高效。

线程池框架Executor

JAVA中的线程池是通过Executor框架实现的,Executor 框架包括类:Executor,Executors,ExecutorService,ThreadPoolExecutor ,Callable和Future、FutureTask的使用等。

Executor: 所有线程池的接口,只有一个方法。

public interface Executor { 
      void execute(Runnable command); 
}

ExecutorService: 增加Executor的行为,是Executor实现类的最直接接口。

Executors: 提供了一系列工厂方法用于创先线程池,返回的线程池都实现了ExecutorService 接口。

ThreadPoolExecutor:线程池的具体实现类,一般用的各种线程池都是基于这个类实现的。

构造方法

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

参数介绍

corePoolSize:线程池的核心线程数,线程池中运行的线程数也永远不会超过 corePoolSize 个,默认情况下可以一直存活。可以通过设置allowCoreThreadTimeOut为True,此时 核心线程数就是0,此

keepAliveTime控制所有线程的超时时间。

maximumPoolSize:线程池允许的最大线程数;

keepAliveTime: 指的是空闲线程结束的超时时间;

unit :是一个枚举,表示 keepAliveTime 的单位;

workQueue:表示存放任务的BlockingQueue<Runnable队列。

BlockingQueue:阻塞队列(BlockingQueue)是java.util.concurrent下的主要用来控制线程同步的工具。如果BlockQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直

BlockingQueue进了东西才会被唤醒。同样,如果BlockingQueue是满的,任何试图往里存东西的操作也会被阻断进入等待状态,直到BlockingQueue里有空间才会被唤醒继续操作。

阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素。具体的实现类有LinkedBlockingQueue,ArrayBlockingQueued等。一般其内部的都是通过Lock和Condition(显示锁(Lock)及Condition的学习与使用)来实现阻塞和唤醒。

线程池的工作过程

线程池刚创建时,里面没有一个线程。任务队列是作为参数传进来的。不过,就算队列里面有任务,线程池也不会马上执行它们。

当调用 execute() 方法添加一个任务时,线程池会做如下判断:

如果正在运行的线程数量小于 corePoolSize,那么马上创建线程运行这个任务;

如果正在运行的线程数量大于或等于 corePoolSize,那么将这个任务放入队列;

如果这时候队列满了,而且正在运行的线程数量小于 maximumPoolSize,那么还是要创建非核心线程立刻运行这个任务;

如果队列满了,而且正在运行的线程数量大于或等于 maximumPoolSize,那么线程池会抛出异常RejectExecutionException。

当一个线程完成任务时,它会从队列中取下一个任务来执行。

当一个线程无事可做,超过一定的时间(keepAliveTime)时,线程池会判断,如果当前运行的线程数大于 corePoolSize,那么这个线程就被停掉。所以线程池的所有任务完成后,它最终会收缩到 corePoolSize 的大小。

线程池的创建和使用

生成线程池采用了工具类Executors的静态方法,以下是几种常见的线程池

SingleThreadExecutor:单个后台线程 (其缓冲队列是无界的)

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

创建一个单线程的线程池。这个线程池只有一个核心线程在工作,也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束,那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

2)FixedThreadPool:只有核心线程的线程池,大小固定 (其缓冲队列是无界的) 。

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

创建固定大小的线程池。每次提交一个任务就创建一个线程,直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变,如果某个线程因为执行异常而结束,那么线程池会补充一个新线程。

3)CachedThreadPool:无界线程池,可以进行自动线程回收。

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

如果线程池的大小超过了处理任务所需要的线程,那么就会回收部分空闲(60秒不执行任务)的线程,当任务数增加时,此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制,线程池大小完全依赖于操作系统(或者说JVM)能够创建的最大线程大小。SynchronousQueue是一个是缓冲区为1的阻塞队列。

4)ScheduledThreadPool:核心线程池固定,大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。

public static ExecutorService newScheduledThreadPool(int corePoolSize) {
       return new ScheduledThreadPool(corePoolSize, Integer.MAX_VALUE, DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
         new DelayedWorkQueue());
}

线程池实现的原理

如果只讲线程池的使用,那这篇博客没有什么大的价值,充其量也就是熟悉Executor相关API的过程。线程池的实现过程没有用到Synchronized关键字,用的都是volatile,Lock和同步(阻塞)队列,Atomic相关类,FutureTask等等,因为后者的性能更优。理解的过程可以很好的学习源码中并发控制的思想。

在ThreadPoolExecutor主要Worker类来控制线程的复用。看下Worker类简化后的代码,这样方便理解:

private final class Worker implements Runnable {
      final Thread thread;
      Runnable firstTask;
      Worker(Runnable firstTask) {
          this.firstTask = firstTask;
          this.thread = getThreadFactory().newThread(this);
      }

     public void run() {
          runWorker(this);
      }

     final void runWorker(Worker w) {
          Runnable task = w.firstTask;
          w.firstTask = null;
          while (task != null || (task = getTask()) != null){
          task.run();
     }
}

Worker是一个Runnable,同时拥有一个thread,这个thread就是要开启的线程,在新建Worker对象时同时新建一个Thread对象,同时将Worker自己作为参数传入TThread,这样当Thread的start()方法调用时,运行的实际上是Worker的run()方法,接着到runWorker()中,有个while循环,一直从getTask()里得到Runnable对象,顺序执行。getTask()又是怎么得到Runnable对象的呢?

private Runnable getTask() {
     if(一些特殊情况) {
        return null;
     }

    Runnable r = workQueue.take();
    return r;
}

这个workQueue就是初始化ThreadPoolExecutor时存放任务的BlockingQueue队列,这个队列里的存放的都是将要执行的Runnable任务。因为BlockingQueue是个阻塞队列,BlockingQueue.take()得到如果是空,则进入等待状态直到BlockingQueue有新的对象被加入时唤醒阻塞的线程。所以一般情况Thread的run()方法就不会结束,而是不断执行从workQueue里的Runnable任务,这就达到了线程复用的原理了。

控制最大并发数

那Runnable是什么时候放入workQueue?Worker又是什么时候创建,Worker里的Thread的又是什么时候调用start()开启新线程来执行Worker的run()方法的呢?有上面的分析看出Worker里的runWorker()执行任务时是一个接一个,串行进行的,那并发是怎么体现的呢?

很容易想到是在execute(Runnable runnable)时会做上面的一些任务。看下execute里是怎么做的。

public void execute(Runnable command) {
      if (command == null)
        throw new NullPointerException();

      int c = ctl.get();
      // 当前线程数 < corePoolSize
      if (workerCountOf(c) < corePoolSize) {
      // 直接启动新的线程。
      if (addWorker(command, true))
         return;
         c = ctl.get();
}
    // 活动线程数 >= corePoolSize
    // runState为RUNNING && 队列未满
    // workQueue.offer(command)表示添加到队列,如果添加成功返回true,否则返回false
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
    // 再次检验是否为RUNNING状态
    // 非RUNNING状态 则从workQueue中移除任务并拒绝
    if (!isRunning(recheck) && remove(command))
        reject(command);// 采用线程池指定的策略拒绝任务
    // 两种情况
    // 1.非RUNNING状态拒绝新的任务
    // 2.队列满了启动新的线程失败(workCount > maximumPoolSize)
    } else if (!addWorker(command, false))
        reject(command);
}

根据代码再来看上面提到的线程池工作过程中的添加任务的情况:

如果正在运行的线程数量小于 corePoolSize,那么马上创建线程运行这个任务;

如果正在运行的线程数量大于或等于 corePoolSize,那么将这个任务放入队列;

如果这时候队列满了,而且正在运行的线程数量小于 maximumPoolSize,那么还是要创建非核心线程立刻运行这个任务;

如果队列满了,而且正在运行的线程数量大于或等于 maximumPoolSize,那么线程池会抛出异常RejectExecutionException。

原文链接:慕课网 http://www.imooc.com/article/19226

参与评论