Thread

简介

现代操作系统调度的最小单元是线程，也叫轻量级进程，在一个进程里可创建多个线程，这些线程都拥有各自的计数器、堆栈和局部变量等属性，并且能够访问共享的内存变量。处理器在这些线程上高速切换，让使用者感觉到这些线程在同时执行。

为什么要使用多线程

• 更多的处理器核心
• 更快的相应时间
• 更好到编程模型

线程的状态

Java线程在运行的生命周期中可能处于6种不同的状态，在给定的一个时刻，线程只能处于其他的一个状态

常用方法介绍

等待与通知

一个线程修改了一个对象的值，而另一个线程感知到了变化，然后进行相应的操作，整个过程开始于一个线程，而最终执行又是另一个线程。前者是生产者,后者是消费者,这种模式隔离了”怎么做”,”做什么”，在功能层面上实现了解耦，体系结构上具备了良好的伸缩性。

等待方的原则:

1. 获取对象的锁
2. 如果条件不满足，那么调用锁的wait()方法，使该线程进入waiting，被通知后依然要检查条件
3. 条件满足则执行对应的逻辑

伪代码:

  synchronized(对象){
    while(条件不满足){
     对象.wait();
  }
  对应的逻辑处理
}

通知方的原则:

1. 获取对象的锁
2. 改变条件
3. 通知所有等待在该对象上的线程

伪代码:

  synchronized(对象){
     改变条件
     对象.notifyAll();
}

实现代码：

public class WaitNotify{
  static boolean flag = true;
  static Object lock = new Object();

 public static void main(String[] args) throws Exception{
           Thread waitThread = new Thread(new Wait(),"WaitThread");
           waitThread.start();
           TimeUtil.SECONDS.sleep(1);
           Thread notifyThread = new Thread(new Notify(),"NotifyThread");
           notifyThread.start();
    }

static class  Wait implements Runnable{
   public void run(){
      //加锁
      synchronized(lock){
       //当条件不满足的时候，进入WAITTING状态，同时释放lock锁
       while(flag){
        System.out.println("flag is true ");
          lock.wait();
          }
        //条件满足
        System.out.println("doSomething");
      }
    }
}
 static class  Notify implements Runnable{
   public void run(){
      //加锁
      synchronized(lock){
         //获取lock的锁，然后进行通知，通知不会释放lock锁
         //直到发出通知的线程执行完毕释放了lock锁，WaitThread线程才能从wait方法返回
         lock.notifyAll();
          System.out.println("flag is false now");
         flag = false;
      }
   }
}

输出内容如下:

flag is true

flag is false now

doSomething

线程池

线程池的应用场景

Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中，合理地使用线程池能够带来3个好处

1. 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

2. 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。

3. 提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。但是，要做到合理利用线程池，必须对其实现原理了如指掌

线程池的实现原理-ThreadPoolExecutor

ThreadPoolExecutor执行execute方法分下面4种情况。

1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。
2. 如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
3. 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（注意，执
   行这一步骤需要获取全局锁）。
4. 如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路，是为了在执行execute()方法时，尽可能地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热之后（当前运行的线程数大于等于corePoolSize），几乎所有的execute()方法调用都是执行步骤2，而步骤2不需要获取全局锁。

executor执行流程(基于JDK7.0)

   public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();//AtomicInteger
       //1.
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //2.
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))//（21）
                reject(command);
            else if (workerCountOf(recheck) == 0)//（22）
                addWorker(null, false);//为什么是false
        }//3.
        else if (!addWorker(command, false))
            reject(command);
    }

1. 首先查看了当前线程池中的线程数量是否小于我们指定的核心线程池的数目，如果是就尝试新建一个线程，把command作为他的第一个任务，并把他们加入到线程池中。但是我们在判断了线程池的数量合法后，调用addWorker(command, true)把线程加入到线程池中时，是多线程并发的，可能会导致加入失败。如果加入成功，则直接返回，若假如失败，则重新获取clt，因为此时clt必发生了变化，否则不会失败，继续往下执行（2）。
2. 通过isRunning(c) 判断如果线程池还在运行，那我们就尝试把当前的command加入到阻塞队列中。加入的过程也是并发的，也可能会出现失败。如果失败在继续执行（3）。加入阻塞队列成功后我们要重新在检查一遍，防止在加入的过程中线程时关闭了或者线程池中没有线程了，全部因为空闲时间超过了我们指定的alivetime被回收了。如果是线程池已经不再是RUNNING状态，则用我们的拒绝策略去丢弃它（21）。如果是线程池没有了线程，那我们新建一个空线程，让他去阻塞队列中去获取任务执行（22）。
3. 如果上面的两步都没有执行成功，那我们此时就需要使用我们指定的最大线程池，来处理它，但是此时也是可能失败的，可能有多个线程执行么，如果失败，就用拒绝策略丢弃该线程。

  private boolean addWorker(Runnable firstTask, boolean core) {
        //（1）循环CAS操作，将线程池中的线程数+1.
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.

            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                //core true代表是往核心线程池中增加线程 false代表往最大线程池中增加线程
                //线程数超标，不能再添加了，直接返回
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS修改clt的值+1，在线程池中为将要添加的线程流出空间，成功退出cas循环，失败继续
                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
            }
        }
        //(2)新建线程，并加入到线程池workers中。
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            //对workers操作要通过加锁来实现
            final ReentrantLock mainLock = this.mainLock;
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
               //细化锁的力度，防止临界区过大，浪费时间
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    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();
                       //将新建的线程加入到线程池中
                        workers.add(w);
                        int s = workers.size();
                        //修正largestPoolSize的值
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //线程添加线程池成功，则开启新创建的线程
                if (workerAdded) {
                    t.start();//(3)
                    workerStarted = true;
                }
            }
        } finally {
            //线程添加线程池失败或者线程start失败，则需要调用addWorkerFailed函数，如果添加成功则需要移除，并回复clt的值
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

如何实现线程复用

线程池中，每一个线程都是一个Worker，Worker是一个内部类，继承了AbstractQueuedSynchronizer。

Worker的run方法会调用runWorker方法，runWorker方法会循环调用getTask获取阻塞队列的任务，达到线程复现的目的

Worker的主要字段就下面三个，代码也比较简单。

        //线程池中正真运行的线程。通过我们指定的线程工厂创建而来
        final Thread thread;
        //线程包装的任务。thread 在run时主要调用了该任务的run方法
        Runnable firstTask;
        //记录当前线程完成的任务数
        volatile long completedTasks;

Worker的构造函数

        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker（写的很清楚）
            this.firstTask = firstTask;
            //利用我们指定的线程工厂创建一个线程，注意，参数是this，也就是在执行thread.run时，正真执行的是我们Woker类的run方法
            this.thread = getThreadFactory().newThread(this);
        }

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                //线程池处于stop状态或者当前线程被中断时，线程池状态是stop状态。但是当前线程没有中断，则发出中断请求
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                //开始执行任务前的Hook，类似回调函数
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        //执行任务
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                    //任务执行后的Hook，类似回调函数
                        afterExecute(task, thrown);
                    }
                } finally {
                //执行完毕后task重置，completedTasks计数器++，解锁
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
           //线程空闲达到我们设定的值时，Worker退出销毁。
            processWorkerExit(w, completedAbruptly);
        }
    }

 private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            //如果线程池处于shutdown状态，并且队列为空，或者线程池处于stop或者terminate状态，在线程池数量-1，返回null，回收线程
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            //标识当前线程在空闲时，是否应该超时回收
            boolean timed;    

            for (;;) {
                int wc = workerCountOf(c);
                //如果allowCoreThreadTimeOut 为ture或者当前线程数量大于核心线程池数目，则需要超时回收
                timed = allowCoreThreadTimeOut || wc > corePoolSize;
                //（1）
                //如果线程数目小于最大线程数目，且不允许超时回收或者未超时，则跳出循环，继续去阻塞队列中取任务（2）
                if (wc <= maximumPoolSize && ! (timedOut && timed))
                    break;
                //如果上面if没有成立，则当前线程数-1，返回null，回收该线程
                if (compareAndDecrementWorkerCount(c))
                    return null;
                //如果上面if没有成立，则CAS修改ctl失败，重读，cas循环重新尝试修改
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }

            （2）
            try {
            //如果允许空闲回收，则调用阻塞队列的poll，否则take，一直等到队列中有可取任务
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                //取到任务，返回任务，否则超时timedOut = true;进入下一个循环,并且在（1）处会不成立，进而进入到cas修改ctl的程序中
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

如何处理异常

当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法，默认的异常处理为AbortPolicy

• ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

    /**
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always.
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

• ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。

   /**
     * A handler for rejected tasks that silently discards the
     * rejected task.
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

• ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）

  /**
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries {@code execute}, unless the executor
     * is shut down, in which case the task is discarded.
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

• ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

 /**
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always.
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

Executor框架

FixedThreadPool

FixedThreadPool被称为可重用固定线程数的线程池

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

FixedThreadPool适用于为了满足资源管理的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器

SingleThreadExecutor

SingleThreadExecutor。下面是Executors提供的，创建使用单个线程的SingleThread-
Executor的API

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

SingleThreadExecutor适用于需要保证顺序地执行各个任务；并且在任意时间点，不会有多
个线程是活动的应用场景

CachedThreadPool

CachedThreadPool。下面是Executors提供的，创建一个会根据需要创建新线程的CachedThreadPool的API

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

CachedThreadPool是大小无界的线程池，适用于执行很多的短期异步任务的小程序，或者是负载较轻的服务器。

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor通常使用工厂类Executors来创建。Executors可以创建2种类型的ScheduledThreadPoolExecutor，如下。

• ScheduledThreadPoolExecutor。包含若干个线程的ScheduledThreadPoolExecutor。
• SingleThreadScheduledExecutor。只包含一个线程的ScheduledThreadPoolExecutor。

 public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }

  public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

ScheduledThreadPoolExecutor适用于需要多个后台线程执行周期任务，同时为了满足资源管理的需求而需要限制后台线程的数量的应用场景

参考资料

《Java并发编程的艺术》

JDK1.7中的ThreadPoolExecutor源码剖析

Java Thread 的使用