云顶娱乐手机官网-云顶娱乐网址

热门关键词: 云顶娱乐手机官网,云顶娱乐网址

Java并发编制程序--线程池

2019-09-23 作者:编辑程序   |   浏览(162)

Executor 接口

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

大家能够看到 Executor 接口特别轻松,就一个void execute(Runnable command)主意,代表提交二个职务。

当然了,Executor 这一个接口只有付诸职务的作用,太轻便了,大家想要更丰硕的效率,比如我们想理解试行结果、大家想知道当前线程池有多少个线程活着、已经实现了稍稍任务等等,那么些都以其一接口的缺乏的地点。接下来大家要介绍的是三翻五次自Executor接口的ExecutorService接口,这一个接口提供了比较充裕的功效,也是大家最常使用到的接口。

线程池状态

成员变量ctl是个Integer的原子变量用来记录线程池状态和线程池线程个数,在那之中Integer类型是叁十二位二进制标示,当中高3位用来代表线程池状态,前边26人用来记录线程池线程个数。

//用来标记线程池状态(高3位),线程个数(低29位)
//默认是RUNNING状态,线程个数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

//线程个数掩码位数
private static final int COUNT_BITS = Integer.SIZE - 3;

//线程最大个数(低29位)00011111111111111111111111111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

//(高3位):11100000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;

//(高3位):00000000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;

//(高3位):00100000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;

//(高3位):01000000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;

//(高3位):01100000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;

// 获取高三位 运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }

//获取低29位 线程个数
private static int workerCountOf(int c)  { return c & CAPACITY; }

//计算ctl新值,线程状态 与 线程个数
private static int ctlOf(int rs, int wc) { return rs | wc; }

线程池状态含义:

  • RUNNING:接受新职务而且管理阻塞队列里的任务
  • SHUTDOWN:拒绝新任务然则管理阻塞队列里的任务
  • STOP:拒绝新任务而且扬弃阻塞队列里的天职相同的时间会中断正在管理的天职
  • TIDYING:全体职务都推行完(包涵阻塞队列之中职分)当前线程池活动线程为0,将在调用terminated方法
  • TERMINATED:终止景况。terminated方法调用达成未来的情事

     *   RUNNING:  Accept new tasks and process queued tasks
     *   SHUTDOWN: Don't accept new tasks, but process queued tasks
     *Java并发编制程序--线程池。   STOP:  Don't accept new tasks, don't process queued tasks,and interrupt in-progress tasks
     *   TIDYING:  All tasks have terminated, workerCount is zero, the thread transitioning to state TIDYING, will run the terminated() hook method
     *   TERMINATED: terminated() has completed

线程池状态调换:

RUNNING -> SHUTDOWN
显式调用shutdown()方法,只怕隐式调用了finalize(),它个中调用了shutdown()方法。

RUNNING or SHUTDOWN)-> STOP
显式 shutdownNow()方法

SHUTDOWN -> TIDYING
当线程池和任务队列都为空的时候

STOP -> TIDYING
当线程池为空的时候

TIDYING -> TERMINATED
当 terminated() hook 方法实践到位时候

总览

下图是 java 线程池几个相关类的两次三番结构:

图片 1图片 2图片 3

先轻便说说这几个一连结构,Executor 位于最顶层,也是最简易的,就一个execute(Runnable runnable) 接口方法定义。

ExecutorService 也是接口,在 Executor 接口的基本功上增添了数不清的接口方法,所以一般的话大家会选拔这些接口。

下一场再下来一层是 AbstractExecutorService,从名字大家就知晓,这是抽象类,这里达成了特别管用的部分主意供子类直接运用,之后我们再细说。

下一场才到大家的基本点部分 ThreadPoolExecutor 类,这些类提供了关于线程池所需的特别丰裕的法力。

线程池中的 BlockingQueue 也是非凡主要的概念,假使线程数达到corePoolSize,我们的各种职分会交到到等候队列中,等待线程池中的线程来取职务并实行。这里的 BlockingQueue 日常大家使用其完毕类 LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue,每一个完结类都有两样的性格,使用情状过后会渐渐深入分析。想要详细询问各个BlockingQueue 的读者,可以参见小编的眼下的一篇对 BlockingQueue 的逐个完成类举行详细剖析的稿子。

1.ThreadPoolExecutor类

java.uitl.concurrent.ThreadPoolExecutor类是线程池中最大旨的一个类,上边大家来看一下ThreadPoolExecutor类的切实可行落实源码(内容听他们讲JDK1.7)。

在ThreadPoolExecutor类中提供了多少个构造方法:

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

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

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

    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承接了AbstractExecutorService类,并提供了多个构造器,事实上,开掘前边多个构造器都是调用的第多个构造器进行的初叶化专门的职业。

下边解释下一下构造器中逐个参数的意思:

  • corePoolSize:线程池大旨线程大小。在开立了线程池后,暗中同意情状下,线程池中并不曾别的线程,而是等待有任务赶来才创造线程去实施任务,除非调用了prestartAllCoreThreads()或然prestartCoreThread()方法,从那2个艺术的名字就足以看到,是预创造线程的野趣,即在平素不任务赶来从前就成立corePoolSize个线程大概叁个线程。暗许景况下,在开创了线程池后,线程池中的线程数为0,当有职务来过后,就能成立一个线程去实践职务,当线程池中的线程数目达到corePoolSize后,就能把达到的天职放到缓存队列当中;
  • maximumPoolSize:线程池最大线程数,表示在线程池中最多能创立多少个线程;
  • keepAliveTime:表示线程未有任务施行时最多维持多长期光阴会终止。私下认可景况下,唯有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起功用,直到线程池中的线程数不高于corePoolSize,即当线程池中的线程数大于corePoolSize时,如若二个线程空闲的年华达到keepAliveTime,则会终止,直到线程池中的线程数不当先corePoolSize。不过倘诺调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不高于corePoolSize时,keepAliveTime参数也会起效果,直到线程池中的线程数为0;
  • unit:参数的时光单位,有7种取值,在提姆eUnit类中有7种静态属性:

    TimeUnit.DAYS;               //天
    TimeUnit.HOURS;             //小时
    TimeUnit.MINUTES;           //分钟
    TimeUnit.SECONDS;           //秒
    TimeUnit.MILLISECONDS;      //毫秒
    TimeUnit.MICROSECONDS;      //微秒
    TimeUnit.NANOSECONDS;       //纳秒
    
  • workQueue:多个梗阻队列,用来积存等待实践的天职;

  • threadFactory:线程工厂,主要用于创造线程;
  • handler:表示当拒绝管理职务时的攻略,有以下多种取值:
    ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 
    ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。 
    ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
    ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务 
    

从ThreadPoolExecutor类能够精通,ThreadPoolExecutor承接了AbstractExecutorService,我们来看一下AbstractExecutorService的兑现:

public abstract class AbstractExecutorService implements ExecutorService {

    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        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;
    }

    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                            boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null)
            throw new NullPointerException();
        int ntasks = tasks.size();
        if (ntasks == 0)
            throw new IllegalArgumentException();
        List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);
        ExecutorCompletionService<T> ecs =
            new ExecutorCompletionService<T>(this);

        // For efficiency, especially in executors with limited
        // parallelism, check to see if previously submitted tasks are
        // done before submitting more of them. This interleaving
        // plus the exception mechanics account for messiness of main
        // loop.

        try {
            // Record exceptions so that if we fail to obtain any
            // result, we can throw the last exception we got.
            ExecutionException ee = null;
            long lastTime = timed ? System.nanoTime() : 0;
            Iterator<? extends Callable<T>> it = tasks.iterator();

            // Start one task for sure; the rest incrementally
            futures.add(ecs.submit(it.next()));
            --ntasks;
            int active = 1;

            for (;;) {
                Future<T> f = ecs.poll();
                if (f == null) {
                    if (ntasks > 0) {
                        --ntasks;
                        futures.add(ecs.submit(it.next()));
                        ++active;
                    }
                    else if (active == 0)
                        break;
                    else if (timed) {
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        long now = System.nanoTime();
                        nanos -= now - lastTime;
                        lastTime = now;
                    }
                    else
                        f = ecs.take();
                }
                if (f != null) {
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }

            if (ee == null)
                ee = new ExecutionException();
            throw ee;

        } finally {
            for (Future<T> f : futures)
                f.cancel(true);
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        try {
            return doInvokeAny(tasks, false, 0);
        } catch (TimeoutException cannotHappen) {
            assert false;
            return null;
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return doInvokeAny(tasks, true, unit.toNanos(timeout));
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                execute(f);
            }
            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    try {
                        f.get();
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures)
                    f.cancel(true);
        }
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        if (tasks == null || unit == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks)
                futures.add(newTaskFor(t));

            long lastTime = System.nanoTime();

            // Interleave time checks and calls to execute in case
            // executor doesn't have any/much parallelism.
            Iterator<Future<T>> it = futures.iterator();
            while (it.hasNext()) {
                execute((Runnable)(it.next()));
                long now = System.nanoTime();
                nanos -= now - lastTime;
                lastTime = now;
                if (nanos <= 0)
                    return futures;
            }

            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    if (nanos <= 0)
                        return futures;
                    try {
                        f.get(nanos, TimeUnit.NANOSECONDS);
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    } catch (TimeoutException toe) {
                        return futures;
                    }
                    long now = System.nanoTime();
                    nanos -= now - lastTime;
                    lastTime = now;
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures)
                    f.cancel(true);
        }
    }

}

AbstractExecutorService是二个抽象类,它实现了ExecutorService接口,而ExecutorService又是持续了Executor接口,它们的主导关系如下:

  • Executor是二个顶层接口,在它个中只申明了八个方法execute(Runnable),再次来到值为void,参数为Runnable类型,用来实行传进去的天职的;
  • ExecutorService接口承接了Executor接口,并声称了部分主意:submit、invokeAll、invokeAny以及shutDown等;
  • 抽象类AbstractExecutor瑟维斯完成了ExecutorService接口,基本实现了ExecutorService中注明的具有办法;
  • ThreadPoolExecutor承继了类AbstractExecutorService,

在ThreadPoolExecutor类中有多少个相当重大的法子:

execute()
submit()
shutdown()
shutdownNow()
  • execute()方法其实是Executor中声称的章程,在ThreadPoolExecutor进行了实际的落成,那些方法是ThreadPoolExecutor的主题措施,通过那么些法子能够向线程池提交五个职分,交由线程池去施行。
  • submit()方法是在ExecutorService中声称的秘诀,在AbstractExecutorService就曾经有了现实的贯彻,在ThreadPoolExecutor中并不曾对其伸开重写,这一个点子也是用来向线程池提交职责的,不过它和execute()方法差异,它亦可回到职分施行的结果,去看submit()方法的贯彻,会意识它实际上依旧调用的execute()方法,只可是它利用了Future来获取义务试行结果。
  • shutdown()和shutdownNow()是用来关闭线程池的。

选取示例

 1 package main.java.Juc; 2  3 import java.util.concurrent.ExecutorService; 4 import java.util.concurrent.Executors; 5  6 class MyRunnable implements Runnable { 7     @Override 8     public void run() { 9         for (int x = 0; x < 100; x++) {10             System.out.println(Thread.currentThread().getName() + ":" + x);11         }12     }13 }14 15 public class TestThreadPool {16     public static void main(String[] args) {17         // 创建一个线程池对象,控制要创建几个线程对象。18         ExecutorService pool = Executors.newFixedThreadPool(2);19 20         // 可以执行Runnable对象或者Callable对象代表的线程21         pool.execute(new MyRunnable;22         pool.execute(new MyRunnable;23 24         //结束线程池25         pool.shutdown();26     }27 }

运转结果:

图片 4

2.深入分析线程池完结原理

ThreadPoolExecutor

大家日常会接纳Executors那一个工具类来火速构造七个线程池,对于初学者来说,这种工具类是很有用的,开采者无需关切太多的内情,只要驾驭本身索要三个线程池,仅仅提供须要的参数就能够了,别的参数都应用小编提供的暗中认可值。

 1 public static ExecutorService newFixedThreadPool(int nThreads) { 2     return new ThreadPoolExecutor(nThreads, nThreads, 3                                   0L, TimeUnit.MILLISECONDS, 4                                   new LinkedBlockingQueue<Runnable>; 5 } 6 public static ExecutorService newCachedThreadPool() { 7     return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 8                                   60L, TimeUnit.SECONDS, 9                                   new SynchronousQueue<Runnable>;10 }

那边先不说有怎么着界别,它们最后都会导向这些构造方法:

 1 public ThreadPoolExecutor(int corePoolSize, 2                           int maximumPoolSize, 3                           long keepAliveTime, 4                           TimeUnit unit, 5                           BlockingQueue<Runnable> workQueue, 6                           ThreadFactory threadFactory, 7                           RejectedExecutionHandler handler) { 8     if (corePoolSize < 0 || 9         maximumPoolSize <= 0 ||10         maximumPoolSize < corePoolSize ||11         keepAliveTime < 0)12         throw new IllegalArgumentException();13     // 这几个参数都是必须要有的14     if (workQueue == null || threadFactory == null || handler == null)15         throw new NullPointerException();16 17     this.corePoolSize = corePoolSize;18     this.maximumPoolSize = maximumPoolSize;19     this.workQueue = workQueue;20     this.keepAliveTime = unit.toNanos(keepAliveTime);21     this.threadFactory = threadFactory;22     this.handler = handler;23 }

上面包车型客车构造方法中列出了大家最必要关注的几本性情了,下边每一个介绍下构造方法中出现的那多少个属性:

  • corePoolSize

*    *线程池中的核心线程数。

  • maximumPoolSize

*    *最大线程数,线程池允许创造的最大线程数。要是当前不通队列满了,且三回九转提交职责,则创制新的线程实施任务,前提是当前线程数稍低于maximumPoolSize;当阻塞队列是无界队列, 则maximumPoolSize则不起功能,因为无法提交至主旨线程池的线程会一直持续地放入workQueue

  • workQueue

    用来保存等待被推行的职务的阻塞队列. 在JDK中提供了如下阻塞队列:

     ArrayBlockingQueue:基于数组结构的有界阻塞队列,按FIFO排序职务;
     LinkedBlockingQuene:基于链表结构的封堵队列,按FIFO排序职责,吞吐量平日要高于ArrayBlockingQuene;
     SynchronousQuene:八个不存款和储蓄成分的鸿沟队列,各类插入操作必得等到另五个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量平时要高于LinkedBlockingQuene;
     priorityBlockingQuene:具有优先级的无界阻塞队列;

    风乐趣的能够看看自家日前关于BlockingQuene的篇章

  • keepAliveTime

    空闲线程的保活时间,如若某线程的空余时间超越这几个值都尚未任务给它做,那么能够被关闭了。注意那些值并不会对负有线程起作用,假诺线程池中的线程数少于等于宗旨线程数 corePoolSize,那么那个线程不会因为空闲太长期而被关闭,当然,也能够通过调用allowCoreThreadTimeOut使基本线程数内的线程也足以被回收;暗许情状下,该参数只在线程数大于corePoolSize时才有用, 超越这些时刻的闲暇线程将被甘休。

  • unit

    keepAliveTime的单位

  • threadFactory

    用于生成线程,一般大家能够用暗许的就足以了。常常,大家得以因此它将大家的线程的名字设置得比较可读一些,如 Message-Thread-1, Message-Thread-2 类似那样。

  • handler

    线程池的饱满攻略,当阻塞队列满了,且并未有空余的干活线程,假诺后续提交任务,必需利用一种政策处理该职分,线程池提供了4种政策:

      AbortPolicy:直接抛出特别,暗中同意计策;
      CallerRunsPolicy:用调用者所在的线程来推行任务;
      DiscardOldestPolicy:放任阻塞队列中靠最前的义务,并举行业前职责;
      DiscardPolicy:直接抛弃任务;
    当然也可以依据使用场景实现RejectedExecutionHandler接口,自定义饱和计策,如记录日志或长久化存款和储蓄无法管理的职责。

而外上边多少个属性外,大家再看看其余重要的性质。

 1 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 2  3 // 这里 COUNT_BITS 设置为 29,意味着前三位用于存放线程状态,后29位用于存放线程数 4 private static final int COUNT_BITS = Integer.SIZE - 3; 5  6 // 000 11111111111111111111111111111 7 // 这里得到的是 29 个 1,也就是说线程池的最大线程数是 2^29-1=536870911 8 // 以我们现在计算机的实际情况,这个数量还是够用的 9 private static final int CAPACITY   = (1 << COUNT_BITS) - 1;10 11 // 我们说了,线程池的状态存放在高 3 位中12 // 运算结果为 111跟29个0:111 0000000000000000000000000000013 private static final int RUNNING    = -1 << COUNT_BITS;14 // 000 0000000000000000000000000000015 private static final int SHUTDOWN   =  0 << COUNT_BITS;16 // 001 0000000000000000000000000000017 private static final int STOP       =  1 << COUNT_BITS;18 // 010 0000000000000000000000000000019 private static final int TIDYING    =  2 << COUNT_BITS;20 // 011 0000000000000000000000000000021 private static final int TERMINATED =  3 << COUNT_BITS;22 23 // 将整数 c 的低 29 位修改为 0,就得到了线程池的状态24 private static int runStateOf(int c)     { return c & ~CAPACITY; }25 // 将整数 c 的高 3 为修改为 0,就得到了线程池中的线程数26 private static int workerCountOf(int c)  { return c & CAPACITY; }27 28 private static int ctlOf(int rs, int wc) { return rs | wc; }29 30 private static boolean runStateLessThan(int c, int s) {31     return c < s;32 }33 34 private static boolean runStateAtLeast(int c, int s) {35     return c >= s;36 }37 38 private static boolean isRunning(int c) {39     return c < SHUTDOWN;40 }

在此间,介绍下线程池中的种种状态和气象变化的改造进度:

  • RUNNING:那一个没什么好说的,那是最健康的情形:接受新的任务,管理等待队列中的义务
  • SHUTDOWN:不接受新的职务交给,但是会再而三管理等待队列中的职分
  • STOP:不接受新的天职交给,不再管理等待队列中的职务,中断正在进行任务的线程
  • TIDYING:全部的职分都销毁了,workCount 为 0。线程池的境况在转变为 TIDYING 状态时,会推行钩子方法 terminated()
  • TERMINATED:terminated() 方法停止后,线程池的境况就能够成为那一个

看了这三种景况的牵线,读者大要也得以猜到十之八九的动静转变了,各类状态的更动进程有以下二种:

  • RUNNING -> SHUTDOWN:当调用了 shutdown() 后,会发出那个景况调换,那也是最关键的
  • (RUNNING or SHUTDOWN) -> STOP:当调用 shutdownNow() 后,会发出这么些情况转变,那下要知道 shutDown() 和 shutDownNow() 的区分了
  • SHUTDOWN -> TIDYING:当职分队列和线程池都清空后,会由 SHUTDOWN 转换为 TIDYING
  • STOP -> TIDYING:当职责队列清空后,产生那几个调换
  • TIDYING -> TERMINATED:这一个前边说了,当 terminated() 方法停止后

除此以外,大家还要看看多少个中间类 Worker,因为 Doug Lea 把线程池中的线程包装成了二个个 Worker,翻译成工人,正是线程池中做任务的线程。所以到此地,大家知晓职责是 Runnable(内部叫 task 或 command),线程是 Worker。

 1 private final class Worker 2     extends AbstractQueuedSynchronizer 3     implements Runnable{ 4     private static final long serialVersionUID = 6138294804551838833L; 5  6     // 这个是真正的线程,任务靠你啦 7     final Thread thread; 8  9     // 前面说了,这里的 Runnable 是任务。为什么叫 firstTask?因为在创建线程的时候,如果同时指定了10     // 这个线程起来以后需要执行的第一个任务,那么第一个任务就是存放在这里的(线程可不止执行这一个任务)11     // 当然了,也可以为 null,这样线程起来了,自己到任务队列(BlockingQueue)中取任务(getTask 方法)就行了12     Runnable firstTask;13 14     // 用于存放此线程完全的任务数,注意了,这里用了 volatile,保证可见性15     volatile long completedTasks;16 17     // Worker 只有这一个构造方法,传入 firstTask,也可以传 null18     Worker(Runnable firstTask) {19         setState; // inhibit interrupts until runWorker20         this.firstTask = firstTask;21         // 调用 ThreadFactory 来创建一个新的线程,这里创建的线程到时候用来执行任务22         this.thread = getThreadFactory().newThread(this);23     }24 25     // 这里调用了外部类的 runWorker 方法26     public void run() {27         runWorker(this);28     }29 30     ...31 }

有了上边的那个基础后,我们总算得以看看 ThreadPoolExecutor 的 execute 方法了,后面源码深入分析的时候也说了,各样艺术都最后凭仗于 execute 方法:

 1 public void execute(Runnable command) { 2     if (command == null) 3         throw new NullPointerException(); 4  5     // 前面说的那个表示 "线程池状态" 和 "线程数" 的整数 6     int c = ctl.get(); 7  8     // 如果当前线程数少于核心线程数,那么直接添加一个 worker 来执行任务, 9     // 创建一个新的线程,并把当前任务 command 作为这个线程的第一个任务(firstTask)10     if (workerCountOf < corePoolSize) {11         // 添加任务成功,那么就结束了。提交任务嘛,线程池已经接受了这个任务,这个方法也就可以返回了12         // 至于执行的结果,到时候会包装到 FutureTask 中。13         // 这里的true代表当前线程数小于corePoolSize,表示以corePoolSize为线程数界限14         if (addWorker(command, true))15             return;16         c = ctl.get();17     }18     // 到这里说明,要么当前线程数大于等于核心线程数,要么刚刚 addWorker 失败了19     // 如果线程池处于 RUNNING 状态,把这个任务添加到任务队列 workQueue 中20     if (isRunning && workQueue.offer {21         int recheck = ctl.get();22         // 如果线程池已不处于 RUNNING 状态,那么移除已经入队的这个任务,并且执行拒绝策略23         if (! isRunning && remove24             reject;25         else if (workerCountOf == 0)26             addWorker(null, false);27     }28     // 如果 workQueue 队列满了,那么进入到这个分支29     // 这里的false代表当前线程数大于corePoolSize,表示以 maximumPoolSize 为界创建新的 worker30     // 如果失败,说明当前线程数已经达到 maximumPoolSize,执行拒绝策略31     else if (!addWorker(command, false))32         reject;33 }

大家得以看看大要的实行流程

图片 5

其一措施充足首要 addWorker(Runnable firstTask, boolean core) 方法,大家看看它是怎么开创新的线程的:

 1 // 第一个参数是准备提交给这个线程执行的任务,之前说了,可以为 null 2 // 第二个参数为 true 代表使用核心线程数 corePoolSize 作为创建线程的界线,也就说创建这个线程的时候, 3 //         如果线程池中的线程总数已经达到 corePoolSize,那么返回false 4 //         如果是 false,代表使用最大线程数 maximumPoolSize 作为界线,线程池中的线程总数已经达到 maximumPoolSize,那么返回false 5 private boolean addWorker(Runnable firstTask, boolean core) { 6     retry: 7     for  { 8         int c = ctl.get(); 9         int rs = runStateOf;10 11         // 如果线程池已关闭,并满足以下条件之一,那么不创建新的 worker:12         // 1. 线程池状态大于 SHUTDOWN,其实也就是 STOP, TIDYING, 或 TERMINATED13         // 2. firstTask != null14         // 3. workQueue.isEmpty()15         if (rs >= SHUTDOWN &&16             ! (rs == SHUTDOWN &&17                firstTask == null &&18                ! workQueue.isEmpty19             return false;20 21         for  {22             int wc = workerCountOf;23             //这里就是通过core参数对当前线程数的判断24             if (wc >= CAPACITY ||25                 wc >= (core ? corePoolSize : maximumPoolSize))26                 return false;27             if (compareAndIncrementWorkerCount28                 break retry;29             c = ctl.get();30             if (runStateOf != rs)31                 continue retry;32             // else CAS failed due to workerCount change; retry inner loop33         }34     }35 36     /* 37      * 到这里,我们认为在当前这个时刻,可以开始创建线程来执行任务了,38      */39 40     // worker 是否已经启动41     boolean workerStarted = false;42     // 是否已将这个 worker 添加到 workers 这个 HashSet 中43     boolean workerAdded = false;44     Worker w = null;45     try {46         final ReentrantLock mainLock = this.mainLock;47         // 把 firstTask 传给 worker 的构造方法48         w = new Worker(firstTask);49         // 取 worker 中的线程对象,之前说了,Worker的构造方法会调用 ThreadFactory 来创建一个新的线程50         final Thread t = w.thread;51         if (t != null) {52             // 这个是整个类的全局锁,因为关闭一个线程池需要这个锁,至少我持有锁的期间,线程池不会被关闭53             mainLock.lock();54             try {55 56                 int c = ctl.get();57                 int rs = runStateOf;58 59                 // 小于 SHUTTDOWN 那就是 RUNNING60                 // 如果等于 SHUTDOWN,前面说了,不接受新的任务,但是会继续执行等待队列中的任务61                 if (rs < SHUTDOWN ||62                     (rs == SHUTDOWN && firstTask == null)) {63                     // worker 里面的 thread 可不能是已经启动的64                     if (t.isAlive65                         throw new IllegalThreadStateException();66                     // 加到 workers 这个 HashSet 中67                     workers.add;68                     int s = workers.size();69                     // largestPoolSize 用于记录 workers 中的个数的最大值70                     // 因为 workers 是不断增加减少的,通过这个值可以知道线程池的大小曾经达到的最大值71                     if (s > largestPoolSize)72                         largestPoolSize = s;73                     workerAdded = true;74                 }75             } finally {76                 mainLock.unlock();77             }78             // 添加成功的话,启动这个线程79             if (workerAdded) {80                 // 启动线程,最重要的就是这里,下面我们会讲解如何执行任务81                 t.start();82                 workerStarted = true;83             }84         }85     } finally {86         // 如果线程没有启动,需要做一些清理工作,如前面 workCount 加了 1,将其减掉87         if (! workerStarted)88             addWorkerFailed;89     }90     // 返回线程是否启动成功91     return workerStarted;92 }

地方第81行代码处已经起步了线程,w = new Worker(firstTask);t = w.thread,大家跟着看看Worker这些类

 1 private final class Worker 2     extends AbstractQueuedSynchronizer 3     implements Runnable{ 4     private static final long serialVersionUID = 6138294804551838833L; 5     final Thread thread; 6     Runnable firstTask; 7     volatile long completedTasks; 8  9     // Worker 只有这一个构造方法,传入 firstTask10     Worker(Runnable firstTask) {11         setState; // inhibit interrupts until runWorker12         this.firstTask = firstTask;13         // 调用 ThreadFactory 来创建一个新的线程,这里创建的线程到时候用来执行任务14         // 我们发现创建线程的时候传入的值是this,我们知道创建线程可以通过继承Runnable的方法,15         // Worker继承了Runnable,并且下面重写了run()方法16         this.thread = getThreadFactory().newThread(this);17     }18 19     // 由上面创建线程时传入的this,上面的thread启动后,会执行这里的run()方法,并且此时runWorker传入的也是this20     public void run() {21         runWorker(this);22     }23 }

持续往下看 runWorker 方法:

 1 // 此方法由 worker 线程启动后调用,这里用一个 while 循环来不断地从等待队列中获取任务并执行 2 // 前面说了,worker 在初始化的时候,可以指定 firstTask,那么第一个任务也就可以不需要从队列中获取 3 final void runWorker { 4     Thread wt = Thread.currentThread(); 5     // 该线程的第一个任务 6     Runnable task = w.firstTask; 7     w.firstTask = null; 8     w.unlock(); // allow interrupts 9     boolean completedAbruptly = true;10     try {11         // 循环调用 getTask 获取任务12         while (task != null || (task = getTask != null) {13             w.lock();          14             // 如果线程池状态大于等于 STOP,那么意味着该线程也要中断15             if ((runStateAtLeast, STOP) ||16                  (Thread.interrupted() &&17                   runStateAtLeast, STOP))) &&18                 !wt.isInterrupted19                 wt.interrupt();20             try {21                 beforeExecute;22                 Throwable thrown = null;23                 try {24                     // 到这里终于可以执行任务了,这里是最重要的,task是什么?是Worker 中的firstTask属性25                     // 也就是上面我们使用示例里面的 new MyRunnable()实例,这里就是真正的执行run方法里面的代码26                     task.run();27                 } catch (RuntimeException x) {28                     thrown = x; throw x;29                 } catch  {30                     thrown = x; throw x;31                 } catch (Throwable x) {32                     thrown = x; throw new Error;33                 } finally {34                     afterExecute(task, thrown);35                 }36             } finally {37                 // 一个任务执行完了,这个线程还可以复用,接着去队列中拉取任务执行38                 // 置空 task,准备 getTask 获取下一个任务39                 task = null;40                 // 累加完成的任务数41                 w.completedTasks++;42                 // 释放掉 worker 的独占锁43                 w.unlock();44             }45         }46         completedAbruptly = false;47     } finally {48         // 如果到这里,需要执行线程关闭:49         // 说明 getTask 返回 null,也就是超过corePoolSize的线程过了超时时间还没有获取到任务,也就是说,这个 worker 的使命结束了,执行关闭50         processWorkerExit(w, completedAbruptly);51     }52 }

咱俩看看 getTask() 是怎么获取职责的

 1 // 此方法有三种可能: 2 // 1. 阻塞直到获取到任务返回。我们知道,默认 corePoolSize 之内的线程是不会被回收的, 3 //      它们会一直等待任务 4 // 2. 超时退出。keepAliveTime 起作用的时候,也就是如果这么多时间内都没有任务,那么应该执行关闭 5 // 3. 如果发生了以下条件,此方法必须返回 null: 6 //    - 池中有大于 maximumPoolSize 个 workers 存在(通过调用 setMaximumPoolSize 进行设置) 7 //    - 线程池处于 SHUTDOWN,而且 workQueue 是空的,前面说了,这种不再接受新的任务 8 //    - 线程池处于 STOP,不仅不接受新的线程,连 workQueue 中的线程也不再执行 9 private Runnable getTask() {10     boolean timedOut = false; // Did the last poll() time out?11 12     retry:13     for  {14         int c = ctl.get();15         int rs = runStateOf;16         // 两种可能17         // 1. rs == SHUTDOWN && workQueue.isEmpty()18         // 2. rs >= STOP19         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty {20             // CAS 操作,减少工作线程数21             decrementWorkerCount();22             return null;23         }24 25         boolean timed;      // Are workers subject to culling?26         for  {27             int wc = workerCountOf;28             // 允许核心线程数内的线程回收,或当前线程数超过了核心线程数,那么有可能发生超时关闭29             timed = allowCoreThreadTimeOut || wc > corePoolSize;30             if (wc <= maximumPoolSize && ! (timedOut && timed))31                 break;32             if (compareAndDecrementWorkerCount33                 return null;34             c = ctl.get();  // Re-read ctl35             // compareAndDecrementWorkerCount 失败,线程池中的线程数发生了改变36             if (runStateOf != rs)37                 continue retry;38             // else CAS failed due to workerCount change; retry inner loop39         }40         // wc <= maximumPoolSize 同时没有超时41         try {42             // 到 workQueue 中获取任务43             // 如果timed=wc > corePoolSize=false,我们知道核心线程数之内的线程永远不会销毁,则执行workQueue.take();我前面文章中讲过,take()方法是阻塞方法,如果队里中有任务则取到任务,如果没有任务,则一直阻塞在这里知道有任务被唤醒。44             //如果timed=wc > corePoolSize=true,这里将执行超时策略,poll(keepAliveTime, TimeUnit.NANOSECONDS)会阻塞keepAliveTime这么长时间,没超时就返回任务,超时则返回null.45             Runnable r = timed ?46                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :47                 workQueue.take();48             if (r != null)49                 return r;50             timedOut = true;51         } catch (InterruptedException retry) {52             // 如果此 worker 发生了中断,采取的方案是重试53             // 解释下为什么会发生中断,这个读者要去看 setMaximumPoolSize 方法,54             // 如果开发者将 maximumPoolSize 调小了,导致其小于当前的 workers 数量,55             // 那么意味着超出的部分线程要被关闭。重新进入 for 循环,自然会有部分线程会返回 null56             timedOut = false;57         }58     }59 }

到这里,基本上也说完了全部工艺流程,读者那年应该回到 execute(Runnable command) 方法,有三种情形会调用 reject 来管理职务,因为遵照符合规律的流水生产线,线程池此时无法承受这一个职分,所以须求实行大家的不肯攻略。接下来,大家说一说 ThreadPoolExecutor 中的拒绝攻略。

1 final void reject(Runnable command) {2     // 执行拒绝策略3     handler.rejectedExecution(command, this);4 }

此间的 handler 大家供给在构造线程池的时候就传来这么些参数,它是 RejectedExecutionHandler 的实例。

RejectedExecutionHandler 在 ThreadPoolExecutor 中有多个已经定义好的兑现类可供我们一贯利用,当然,我们也足以落成团结的国策,可是貌似也未尝须求。

 1 // 只要线程池没有被关闭,那么由提交任务的线程自己来执行这个任务。 2 public static class CallerRunsPolicy implements RejectedExecutionHandler { 3     public CallerRunsPolicy() { } 4     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 5         if (!e.isShutdown { 6             r.run(); 7         } 8     } 9 }10 11 // 不管怎样,直接抛出 RejectedExecutionException 异常12 // 这个是默认的策略,如果我们构造线程池的时候不传相应的 handler 的话,那就会指定使用这个13 public static class AbortPolicy implements RejectedExecutionHandler {14     public AbortPolicy() { }15     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {16         throw new RejectedExecutionException("Task " + r.toString() +17                                              " rejected from " +18                                              e.toString;19     }20 }21 22 // 不做任何处理,直接忽略掉这个任务23 public static class DiscardPolicy implements RejectedExecutionHandler {24     public DiscardPolicy() { }25     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {26     }27 }28 29 // 这个相对霸道一点,如果线程池没有被关闭的话,30 // 把队列队头的任务(也就是等待了最长时间的)直接扔掉,然后提交这个任务到等待队列中31 public static class DiscardOldestPolicy implements RejectedExecutionHandler {32     public DiscardOldestPolicy() { }33     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {34         if (!e.isShutdown {35             e.getQueue;36             e.execute;37         }38     }39 }

到此地,ThreadPoolExecutor 算是深入分析得大致了

职务运营

ThreadPoolExecutor重要变量,对应定义可查阅克罗地亚共和国(Republic of Croatia)语:

     
     /**
       * The queue used for holding tasks and handing off to worker
       * threads.  We do not require that workQueue.poll() returning
       * null necessarily means that workQueue.isEmpty(), so rely
       * solely on isEmpty to see if the queue is empty (which we must
       * do for example when deciding whether to transition from
       * SHUTDOWN to TIDYING).  This accommodates special-purpose
       * queues such as DelayQueues for which poll() is allowed to
       * return null even if it may later return non-null when delays
       * expire.
       */
     private final BlockingQueue<Runnable> workQueue;

        /**
         * Lock held on access to workers set and related bookkeeping.
         * While we could use a concurrent set of some sort, it turns out
         * to be generally preferable to use a lock. Among the reasons is
         * that this serializes interruptIdleWorkers, which avoids
         * unnecessary interrupt storms, especially during shutdown.
         * Otherwise exiting threads would concurrently interrupt those
         * that have not yet interrupted. It also simplifies some of the
         * associated statistics bookkeeping of largestPoolSize etc. We
         * also hold mainLock on shutdown and shutdownNow, for the sake of
         * ensuring workers set is stable while separately checking
         * permission to interrupt and actually interrupting.
         */
        private final ReentrantLock mainLock = new ReentrantLock();

        /**
         * Set containing all worker threads in pool. Accessed only when
         * holding mainLock.
         */
        private final HashSet<Worker> workers = new HashSet<Worker>();

        /**
         * Wait condition to support awaitTermination
         */
        private final Condition termination = mainLock.newCondition();

        /**
         * Tracks largest attained pool size. Accessed only under
         * mainLock.
         */
        private int largestPoolSize;

        /**
         * Counter for completed tasks. Updated only on termination of
         * worker threads. Accessed only under mainLock.
         */
        private long completedTaskCount;

        /**
         * Factory for new threads. All threads are created using this
         * factory (via method addWorker).  All callers must be prepared
         * for addWorker to fail, which may reflect a system or user's
         * policy limiting the number of threads.  Even though it is not
         * treated as an error, failure to create threads may result in
         * new tasks being rejected or existing ones remaining stuck in
         * the queue.
         *
         * We go further and preserve pool invariants even in the face of
         * errors such as OutOfMemoryError, that might be thrown while
         * trying to create threads.  Such errors are rather common due to
         * the need to allocate a native stack in Thread#start, and users
         * will want to perform clean pool shutdown to clean up.  There
         * will likely be enough memory available for the cleanup code to
         * complete without encountering yet another OutOfMemoryError.
         */
        private volatile ThreadFactory threadFactory;

        /**
         * Handler called when saturated or shutdown in execute.
         */
        private volatile RejectedExecutionHandler handler;

        /**
         * Timeout in nanoseconds for idle threads waiting for work.
         * Threads use this timeout when there are more than corePoolSize
         * present or if allowCoreThreadTimeOut. Otherwise they wait
         * forever for new work.
         */
        private volatile long keepAliveTime;

        /**
         * If false (default), core threads stay alive even when idle.
         * If true, core threads use keepAliveTime to time out waiting
         * for work.
         */
        private volatile boolean allowCoreThreadTimeOut;

        /**
         * Core pool size is the minimum number of workers to keep alive
         * (and not allow to time out etc) unless allowCoreThreadTimeOut
         * is set, in which case the minimum is zero.
         */
        private volatile int corePoolSize;

        /**
         * Maximum pool size. Note that the actual maximum is internally
         * bounded by CAPACITY.
         */
        private volatile int maximumPoolSize;

在ThreadPoolExecutor类中,最基本的职务交给方法是execute()方法,我们看下execute()方法源码:

 /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get(); //获取当前线程池的状态+线程个数变量

     //当前线程池线程个数是否小于corePoolSize,小于则开启新线程运行
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }

     //如果线程池处于RUNNING状态,则添加任务到阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();

       //如果当前线程池状态不是RUNNING则从队列删除任务,并执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
       //否者如果当前线程池线程空,则添加一个线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
     //如果队列满了,则新增线程,如果线程个数>maximumPoolSize则执行拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

由代码可观看,该措施首要调用addWorker方法,其源码如下:

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            // 等价于s >= SHUTDOWN &&(rs != SHUTDOWN ||firstTask != null || workQueue.isEmpty())
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            //循环增加线程个数
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;

                //cas增加线程个数
                if (compareAndIncrementWorkerCount(c))
                    break retry;

              //cas失败,则查看线程池状态是否变化,变化则跳到外层循环重试重新获取线程池状态,否者内层循环。
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        //ctl更新成功,新增worker
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            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();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }

                //添加成功,启动线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

创设worker成功,Worker类实现啦Runnable接口,大家接下去看下其run方法:

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
     Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        his.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
     }

   /** Delegates main run loop to outer runWorker  */
     public void run() {
       runWorker(this);
     }   

   /**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     *
     * 1. We may start out with an initial task, in which case we
     * don't need to get the first one. Otherwise, as long as pool is
     * running, we get tasks from getTask. If it returns null then the
     * worker exits due to changed pool state or configuration
     * parameters.  Other exits result from exception throws in
     * external code, in which case completedAbruptly holds, which
     * usually leads processWorkerExit to replace this thread.
     *
     * 2. Before running any task, the lock is acquired to prevent
     * other pool interrupts while the task is executing, and
     * clearInterruptsForTaskRun called to ensure that unless pool is
     * stopping, this thread does not have its interrupt set.
     *
     * 3. Each task run is preceded by a call to beforeExecute, which
     * might throw an exception, in which case we cause thread to die
     * (breaking loop with completedAbruptly true) without processing
     * the task.
     *
     * 4. Assuming beforeExecute completes normally, we run the task,
     * gathering any of its thrown exceptions to send to
     * afterExecute. We separately handle RuntimeException, Error
     * (both of which the specs guarantee that we trap) and arbitrary
     * Throwables.  Because we cannot rethrow Throwables within
     * Runnable.run, we wrap them within Errors on the way out (to the
     * thread's UncaughtExceptionHandler).  Any thrown exception also
     * conservatively causes thread to die.
     *
     * 5. After task.run completes, we call afterExecute, which may
     * also throw an exception, which will also cause thread to
     * die. According to JLS Sec 14.20, this exception is the one that
     * will be in effect even if task.run throws.
     *
     * The net effect of the exception mechanics is that afterExecute
     * and the thread's UncaughtExceptionHandler have as accurate
     * information as we can provide about any problems encountered by
     * user code.
     *
     * @param w the worker
     */
    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();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    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 {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }       

始建worker成功,循环从绿灯队列获取task,若赢得task==null,循环截止,移除该工作线程,上面大家看下getTask方法:

 /**
     * Performs blocking or timed wait for a task, depending on
     * current configuration settings, or returns null if this worker
     * must exit because of any of:
     * 1. There are more than maximumPoolSize workers (due to
     *    a call to setMaximumPoolSize).
     * 2. The pool is stopped.
     * 3. The pool is shutdown and the queue is empty.
     * 4. This worker timed out waiting for a task, and timed-out
     *    workers are subject to termination (that is,
     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
     *    both before and after the timed wait.
     *
     * @return task, or null if the worker must exit, in which case
     *         workerCount is decremented
     */
    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

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

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            boolean timed;      // Are workers subject to culling?

            for (;;) {
                int wc = workerCountOf(c);
                timed = allowCoreThreadTimeOut || wc > corePoolSize;

                if (wc <= maximumPoolSize && ! (timedOut && timed))
                    break;
                //工作线程数量减1;runTask循环结束,执行processWorkerExit(w, completedAbruptly),移除工作线程
                if (compareAndDecrementWorkerCount(c))
                    return null;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }

            try {
                //timed=true,阻塞等待一段时间,若取到task==null,则移除该worker;timed=false:一直阻塞等待
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

 部分内容来自互联网,仅供仿照效法

AbstractExecutorService

AbstractExecutorService 抽象类派生自 ExecutorService接口,然后在其基础上贯彻了多少个实用的措施,那一个办法提供给子类举行调用。

以此抽象类完结了ExecutorService 中的submit 方法,newTaskFor 方法用于将职分包装成 FutureTask。定义于最上层接口 Executor中的void execute(Runnable command)出于不要求获得结果,不会议及展览开 FutureTask 的卷入。

 1 public abstract class AbstractExecutorService implements ExecutorService { 2  3     // RunnableFuture 是用于获取执行结果的,我们常用它的子类 FutureTask 4     // 下面两个 newTaskFor 方法用于将我们的任务包装成 FutureTask 提交到线程池中执行 5     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { 6         return new FutureTask<T>(runnable, value); 7     } 8  9     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {10         return new FutureTask<T>;11     }12 13     // 提交任务14     public Future<?> submit(Runnable task) {15         if (task == null) throw new NullPointerException();16         // 1. 将任务包装成 FutureTask17         RunnableFuture<Void> ftask = newTaskFor(task, null);18         // 2. 交给执行器执行,execute 方法由具体的子类来实现19         // 前面也说了,FutureTask 间接实现了Runnable 接口。20         execute;21         return ftask;22     }23 24     public <T> Future<T> submit(Runnable task, T result) {25         if (task == null) throw new NullPointerException();26         // 1. 将任务包装成 FutureTask27         RunnableFuture<T> ftask = newTaskFor(task, result);28         // 2. 交给执行器执行29         execute;30         return ftask;31     }32 33     public <T> Future<T> submit(Callable<T> task) {34         if (task == null) throw new NullPointerException();35         // 1. 将任务包装成 FutureTask36         RunnableFuture<T> ftask = newTaskFor;37         // 2. 交给执行器执行38         execute;39         return ftask;40     }41 }

到这里,大家开掘,那几个抽象类包装了有的中坚的法子,可是submit等艺术,它们都尚未真的开启线程来推行职责,它们都只是在点子内部调用了 execute 方法,所以最首要的 execute(Runnable runnable) 方法还没出现,这里大家要说的就是 ThreadPoolExecutor 类了。

那篇博客深入分析 Java 中线程池的贯彻。

史上最清晰的线程池源码剖判

ExecutorService

那么大家大概初略地来看一下以此接口中都有何方法:

 1 public interface ExecutorService extends Executor { 2  3     // 关闭线程池,已提交的任务继续执行,不接受继续提交新任务 4     void shutdown(); 5  6     // 关闭线程池,尝试停止正在执行的所有任务,不接受继续提交新任务 7     // 它和前面的方法相比,加了一个单词“now”,区别在于它会去停止当前正在进行的任务 8     List<Runnable> shutdownNow(); 9 10     // 线程池是否已关闭11     boolean isShutdown();12 13     // 如果调用了 shutdown() 或 shutdownNow() 方法后,所有任务结束了,那么返回true14     // 这个方法必须在调用shutdown或shutdownNow方法之后调用才会返回true15     boolean isTerminated();16 17     // 等待所有任务完成,并设置超时时间18     // 我们这么理解,实际应用中是,先调用 shutdown 或 shutdownNow,19     // 然后再调这个方法等待所有的线程真正地完成,返回值意味着有没有超时20     boolean awaitTermination(long timeout, TimeUnit unit)21             throws InterruptedException;22 23     // 提交一个 Callable 任务24     <T> Future<T> submit(Callable<T> task);25 26     // 提交一个 Runnable 任务,第二个参数将会放到 Future 中,作为返回值,27     // 因为 Runnable 的 run 方法本身并不返回任何东西28     <T> Future<T> submit(Runnable task, T result);29 30     // 提交一个 Runnable 任务31     Future<?> submit(Runnable task);32     33     ......34 }

这几个方法都很好掌握,三个大致的线程池主要便是这一个成效,能交到职务,能获得结果,能关闭线程池,那也是为啥我们日常用那几个接口的原因。

总结

作者们简要回看下线程创立的流程

  1. 即便当前线程数少于 corePoolSize,那么提交职分的时候创立贰个新的线程,并由那几个线程实践那一个职分;
  2. 假如当前线程数已经高达 corePoolSize,那么将提交的义务增添到行列中,等待线程池中的线程去队列中取职分;
  3. 举个例子队列已满,那么成立新的线程来实行职责,需求保险池中的线程数不会超过maximumPoolSize,要是此时线程数超过了 maximumPoolSize,那么施行拒绝战略。

引入博客

  

鼎鼎大名的线程池。不必要多说!!!!!

本文由云顶娱乐手机官网发布于编辑程序,转载请注明出处:Java并发编制程序--线程池

关键词: