Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

线程池示例

在分析线程池之前,先看一个简单的线程池示例。

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class ThreadPoolDemo1 {

  public static void main(String[] args) {
    // 创建一个可重用固定线程数的线程池
    ExecutorService pool = Executors.newFixedThreadPool(2);
    // 创建实现了Runnable接口对象,Thread对象当然也实现了Runnable接口
    Thread ta = new MyThread();
    Thread tb = new MyThread();
    Thread tc = new MyThread();
    Thread td = new MyThread();
    Thread te = new MyThread();
    // 将线程放入池中进行执行
    pool.execute(ta);
    pool.execute(tb);
    pool.execute(tc);
    pool.execute(td);
    pool.execute(te);
    // 关闭线程池
    pool.shutdown();
  }
}

class MyThread extends Thread {

  @Override
  public void run() {
    System.out.println(Thread.currentThread().getName()+ " is running.");
  }
}

运行结果:

pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.

示例中,包括了线程池的创建,将任务添加到线程池中,关闭线程池这3个主要的步骤。稍后,我们会从这3个方面来分析ThreadPoolExecutor。

线程池源码分析

(一) 创建“线程池”

下面以newFixedThreadPool()介绍线程池的创建过程。

1. newFixedThreadPool()

newFixedThreadPool()在Executors.java中定义,源码如下:

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

说明:newFixedThreadPool(int nThreads)的作用是创建一个线程池,线程池的容量是nThreads。
         newFixedThreadPool()在调用ThreadPoolExecutor()时,会传递一个LinkedBlockingQueue()对象,而LinkedBlockingQueue是单向链表实现的阻塞队列。在线程池中,就是通过该阻塞队列来实现"当线程池中任务数量超过允许的任务数量时,部分任务会阻塞等待"。
关于LinkedBlockingQueue的实现细节,读者可以参考"Java多线程系列--“JUC集合”08之 LinkedBlockingQueue"。

2. ThreadPoolExecutor()

ThreadPoolExecutor()在ThreadPoolExecutor.java中定义,源码如下:

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

说明:该函数实际上是调用ThreadPoolExecutor的另外一个构造函数。该函数的源码如下:

public ThreadPoolExecutor(int corePoolSize,
             int maximumPoolSize,
             long keepAliveTime,
             TimeUnit unit,
             BlockingQueue<Runnable> workQueue,
             ThreadFactory threadFactory,
             RejectedExecutionHandler handler) {
  if (corePoolSize < 0 ||
    maximumPoolSize <= 0 ||
    maximumPoolSize < corePoolSize ||
    keepAliveTime < 0)
    throw new IllegalArgumentException();
  if (workQueue == null || threadFactory == null || handler == null)
    throw new NullPointerException();
  // 核心池大小
  this.corePoolSize = corePoolSize;
  // 最大池大小
  this.maximumPoolSize = maximumPoolSize;
  // 线程池的等待队列
  this.workQueue = workQueue;
  this.keepAliveTime = unit.toNanos(keepAliveTime);
  // 线程工厂对象
  this.threadFactory = threadFactory;
  // 拒绝策略的句柄
  this.handler = handler;
}

说明:在ThreadPoolExecutor()的构造函数中,进行的是初始化工作。
corePoolSize, maximumPoolSize, unit, keepAliveTime和workQueue这些变量的值是已知的,它们都是通过newFixedThreadPool()传递而来。下面看看threadFactory和handler对象。

2.1 ThreadFactory

线程池中的ThreadFactory是一个线程工厂,线程池创建线程都是通过线程工厂对象(threadFactory)来完成的。
上面所说的threadFactory对象,是通过 Executors.defaultThreadFactory()返回的。Executors.java中的defaultThreadFactory()源码如下:

public static ThreadFactory defaultThreadFactory() {
  return new DefaultThreadFactory();
}

defaultThreadFactory()返回DefaultThreadFactory对象。Executors.java中的DefaultThreadFactory()源码如下:

 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-";
  }

  // 提供创建线程的API。
  public Thread newThread(Runnable r) {
    // 线程对应的任务是Runnable对象r
    Thread t = new Thread(group, r,
               namePrefix + threadNumber.getAndIncrement(),
               0);
    // 设为“非守护线程”
    if (t.isDaemon())
      t.setDaemon(false);
    // 将优先级设为“Thread.NORM_PRIORITY”
    if (t.getPriority() != Thread.NORM_PRIORITY)
      t.setPriority(Thread.NORM_PRIORITY);
    return t;
  }
}

说明:ThreadFactory的作用就是提供创建线程的功能的线程工厂。
         它是通过newThread()提供创建线程功能的,下面简单说说newThread()。newThread()创建的线程对应的任务是Runnable对象,它创建的线程都是“非守护线程”而且“线程优先级都是Thread.NORM_PRIORITY”。

2.2 RejectedExecutionHandler

handler是ThreadPoolExecutor中拒绝策略的处理句柄。所谓拒绝策略,是指将任务添加到线程池中时,线程池拒绝该任务所采取的相应策略。
线程池默认会采用的是defaultHandler策略,即AbortPolicy策略。在AbortPolicy策略中,线程池拒绝任务时会抛出异常!
defaultHandler的定义如下:

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
AbortPolicy的源码如下:

public static class AbortPolicy implements RejectedExecutionHandler {
  public AbortPolicy() { }

  // 抛出异常
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
                       " rejected from " +
                       e.toString());
  }
}

(二) 添加任务到“线程池”

1. execute()

execute()定义在ThreadPoolExecutor.java中,源码如下:

public void execute(Runnable command) {
  // 如果任务为null,则抛出异常。
  if (command == null)
    throw new NullPointerException();
  // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
  int c = ctl.get();
  // 当线程池中的任务数量 < "核心池大小"时,即线程池中少于corePoolSize个任务。
  // 则通过addWorker(command, true)新建一个线程,并将任务(command)添加到该线程中;然后,启动该线程从而执行任务。
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;
    c = ctl.get();
  }
  // 当线程池中的任务数量 >= "核心池大小"时,
  // 而且,"线程池处于允许状态"时,则尝试将任务添加到阻塞队列中。
  if (isRunning(c) && workQueue.offer(command)) {
    // 再次确认“线程池状态”,若线程池异常终止了,则删除任务;然后通过reject()执行相应的拒绝策略的内容。
    int recheck = ctl.get();
    if (! isRunning(recheck) && remove(command))
      reject(command);
    // 否则,如果"线程池中任务数量"为0,则通过addWorker(null, false)尝试新建一个线程,新建线程对应的任务为null。
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  // 通过addWorker(command, false)新建一个线程,并将任务(command)添加到该线程中;然后,启动该线程从而执行任务。
  // 如果addWorker(command, false)执行失败,则通过reject()执行相应的拒绝策略的内容。
  else if (!addWorker(command, false))
    reject(command);
}

说明:execute()的作用是将任务添加到线程池中执行。它会分为3种情况进行处理:
        情况1 -- 如果"线程池中任务数量" < "核心池大小"时,即线程池中少于corePoolSize个任务;此时就新建一个线程,并将该任务添加到线程中进行执行。
        情况2 -- 如果"线程池中任务数量" >= "核心池大小",并且"线程池是允许状态";此时,则将任务添加到阻塞队列中阻塞等待。在该情况下,会再次确认"线程池的状态",如果"第2次读到的线程池状态"和"第1次读到的线程池状态"不同,则从阻塞队列中删除该任务。
        情况3 -- 非以上两种情况。在这种情况下,尝试新建一个线程,并将该任务添加到线程中进行执行。如果执行失败,则通过reject()拒绝该任务。

2. addWorker()

addWorker()的源码如下:

private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  // 更新"线程池状态和计数"标记,即更新ctl。
  for (;;) {
    // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
    int c = ctl.get();
    // 获取线程池状态。
    int rs = runStateOf(c);

    // 有效性检查
    if (rs >= SHUTDOWN &&
      ! (rs == SHUTDOWN &&
        firstTask == null &&
        ! workQueue.isEmpty()))
      return false;

    for (;;) {
      // 获取线程池中任务的数量。
      int wc = workerCountOf(c);
      // 如果"线程池中任务的数量"超过限制,则返回false。
      if (wc >= CAPACITY ||
        wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      // 通过CAS函数将c的值+1。操作失败的话,则退出循环。
      if (compareAndIncrementWorkerCount(c))
        break retry;
      c = ctl.get(); // Re-read ctl
      // 检查"线程池状态",如果与之前的状态不同,则从retry重新开始。
      if (runStateOf(c) != rs)
        continue retry;
      // else CAS failed due to workerCount change; retry inner loop
    }
  }

  boolean workerStarted = false;
  boolean workerAdded = false;
  Worker w = null;
  // 添加任务到线程池,并启动任务所在的线程。
  try {
    final ReentrantLock mainLock = this.mainLock;
    // 新建Worker,并且指定firstTask为Worker的第一个任务。
    w = new Worker(firstTask);
    // 获取Worker对应的线程。
    final Thread t = w.thread;
    if (t != null) {
      // 获取锁
      mainLock.lock();
      try {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 再次确认"线程池状态"
        if (rs < SHUTDOWN ||
          (rs == SHUTDOWN && firstTask == null)) {
          if (t.isAlive()) // precheck that t is startable
            throw new IllegalThreadStateException();
          // 将Worker对象(w)添加到"线程池的Worker集合(workers)"中
          workers.add(w);
          // 更新largestPoolSize
          int s = workers.size();
          if (s > largestPoolSize)
            largestPoolSize = s;
          workerAdded = true;
        }
      } finally {
        // 释放锁
        mainLock.unlock();
      }
      // 如果"成功将任务添加到线程池"中,则启动任务所在的线程。
      if (workerAdded) {
        t.start();
        workerStarted = true;
      }
    }
  } finally {
    if (! workerStarted)
      addWorkerFailed(w);
  }
  // 返回任务是否启动。
  return workerStarted;
}

说明:
    addWorker(Runnable firstTask, boolean core) 的作用是将任务(firstTask)添加到线程池中,并启动该任务。
    core为true的话,则以corePoolSize为界限,若"线程池中已有任务数量>=corePoolSize",则返回false;core为false的话,则以maximumPoolSize为界限,若"线程池中已有任务数量>=maximumPoolSize",则返回false。
    addWorker()会先通过for循环不断尝试更新ctl状态,ctl记录了"线程池中任务数量和线程池状态"。
    更新成功之后,再通过try模块来将任务添加到线程池中,并启动任务所在的线程。

从addWorker()中,我们能清晰的发现:线程池在添加任务时,会创建任务对应的Worker对象;而一个Workder对象包含一个Thread对象。(01) 通过将Worker对象添加到"线程的workers集合"中,从而实现将任务添加到线程池中。 (02) 通过启动Worker对应的Thread线程,则执行该任务。

3. submit()

补充说明一点,submit()实际上也是通过调用execute()实现的,源码如下:

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

 (三) 关闭“线程池”

shutdown()的源码如下:

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  // 获取锁
  mainLock.lock();
  try {
    // 检查终止线程池的“线程”是否有权限。
    checkShutdownAccess();
    // 设置线程池的状态为关闭状态。
    advanceRunState(SHUTDOWN);
    // 中断线程池中空闲的线程。
    interruptIdleWorkers();
    // 钩子函数,在ThreadPoolExecutor中没有任何动作。
    onShutdown(); // hook for ScheduledThreadPoolExecutor
  } finally {
    // 释放锁
    mainLock.unlock();
  }
  // 尝试终止线程池
  tryTerminate();
}

说明:shutdown()的作用是关闭线程池。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间: 2017-06-12

Java concurrency线程池之线程池原理(一)_动力节点Java学院整理

ThreadPoolExecutor简介 ThreadPoolExecutor是线程池类.对于线程池,可以通俗的将它理解为"存放一定数量线程的一个线程集合.线程池允许若个线程同时允许,允许同时运行的线程数量就是线程池的容量:当添加的到线程池中的线程超过它的容量时,会有一部分线程阻塞等待.线程池会通过相应的调度策略和拒绝策略,对添加到线程池中的线程进行管理." ThreadPoolExecutor数据结构 ThreadPoolExecutor的数据结构如下图所示: 各个数据在Thread

Java concurrency线程池之线程池原理(四)_动力节点Java学院整理

拒绝策略介绍 线程池的拒绝策略,是指当任务添加到线程池中被拒绝,而采取的处理措施. 当任务添加到线程池中之所以被拒绝,可能是由于:第一,线程池异常关闭.第二,任务数量超过线程池的最大限制. 线程池共包括4种拒绝策略,它们分别是:AbortPolicy, CallerRunsPolicy, DiscardOldestPolicy和DiscardPolicy. AbortPolicy         -- 当任务添加到线程池中被拒绝时,它将抛出 RejectedExecutionException

Java concurrency线程池之线程池原理(三)_动力节点Java学院整理

线程有5种状态:新建状态,就绪状态,运行状态,阻塞状态,死亡状态.线程池也有5种状态:然而,线程池不同于线程,线程池的5种状态是:Running, SHUTDOWN, STOP, TIDYING, TERMINATED. 线程池状态定义代码如下: private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3

Java中的HashSet详解和使用示例_动力节点Java学院整理

第1部分 HashSet介绍 HashSet 简介 HashSet 是一个没有重复元素的集合. 它是由HashMap实现的,不保证元素的顺序,而且HashSet允许使用 null 元素. HashSet是非同步的.如果多个线程同时访问一个哈希 set,而其中至少一个线程修改了该 set,那么它必须 保持外部同步.这通常是通过对自然封装该 set 的对象执行同步操作来完成的.如果不存在这样的对象,则应该使用 Collections.synchronizedSet 方法来"包装" set.

Java 中的HashMap详解和使用示例_动力节点Java学院整理

第1部分 HashMap介绍 HashMap简介 HashMap 是一个散列表,它存储的内容是键值对(key-value)映射. HashMap 继承于AbstractMap,实现了Map.Cloneable.java.io.Serializable接口. HashMap 的实现不是同步的,这意味着它不是线程安全的.它的key.value都可以为null.此外,HashMap中的映射不是有序的. HashMap 的实例有两个参数影响其性能:"初始容量" 和 "加载因子&quo

Java中List Set和Map之间的区别_动力节点Java学院整理

Java集合的主要分为三种类型: • Set(集) • List(列表) • Map(映射) 要深入理解集合首先要了解下我们熟悉的数组: 数组是大小固定的,并且同一个数组只能存放类型一样的数据(基本类型/引用类型),而JAVA集合可以存储和操作数目不固定的一组数据. 所有的JAVA集合都位于 java.util包中! JAVA集合只能存放引用类型的的数据,不能存放基本数据类型. 世间上本来没有集合,(只有数组参考C语言)但有人想要,所以有了集合 有人想有可以自动扩展的数组,所以有了List 有的

Java线程的生命周期和状态控制_动力节点Java学院整理

一.线程的生命周期 线程状态转换图: 1.新建状态 用new关键字和Thread类或其子类建立一个线程对象后,该线程对象就处于新生状态.处于新生状态的线程有自己的内存空间,通过调用start方法进入就绪状态(runnable). 注意:不能对已经启动的线程再次调用start()方法,否则会出现Javalang.IllegalThreadStateException异常. 2.就绪状态 处于就绪状态的线程已经具备了运行条件,但还没有分配到CPU,处于线程就绪队列(尽管是采用队列形式,事实上,把它称

Sql注入原理简介_动力节点Java学院整理

一.什么是sql注入呢? 所谓SQL注入,就是通过把SQL命令插入到Web表单递交或输入域名或页面请求的查询字符串,最终达到欺骗服务器执行恶意的SQL命令,比如先前的很多影视网站泄露VIP会员密码大多就是通过WEB表单递交查询字符暴出的,这类表单特别容易受到SQL注入式攻击.当应用程序使用输入内容来构造动态sql语句以访问数据库时,会发生sql注入攻击.如果代码使用存储过程,而这些存储过程作为包含未筛选的用户输入的字符串来传递,也会发生sql注入. 黑客通过SQL注入攻击可以拿到网站数据库的访问

Java HelloWorld原理分析_动力节点Java学院整理

我们初学java的第一个程序是"hello world" public class HelloWorld { public static void main(String[] args) { System.out.println("hello world"); } } 上面程序到底是怎么在屏幕上输出"hello world"的呢?这就是本来要讲解的内容,即System.out.println("hello world")的原理