java线程池的初探

Posted 2020-11-07 张小明儿java

问题来源

　　发现学习很多技术都提到了线程池的技术，自己的线程池方面没有仔细研究过，现在看了点东西来这里总结下，最近发现写博客是一个很好的锻炼自己并且将学到的东西更加理解的一个方式。

问题探究

　　java的多线程技术应用很广，但凡是请求大的应用都会用到，但是线程是一个稀缺资源不能无限的创建，即使可以创建很多个线程，线程之间的切换也是十分浪费时间的，所以java推出了线程池概念，

线程池就是将一些线程资源放进一个池子中，当有请求的时候在池子中找到一个空闲的线程进行执行，这样我们可以高效的利用线程资源，避免了线程资源的浪费。

线程池

　使用方式

public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService threadPool = Executors.newFixedThreadPool(4);
        for (int i = 0;i<100;i++) {
            //向线程池提交一个任务，交由线程池去执行
            //threadPool.execute(new PreRun());
            //这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，
            //去看submit()方法的 实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果
            Future<Object> f = (Future<Object>)threadPool.submit(new PreRun2());
            System.out.println(f.get());
        }
        //当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。
        //一般情况下会和shutdown方法组合使用
        threadPool.awaitTermination(1200, TimeUnit.MILLISECONDS);
        //将停止接受新的任务, 同时等待已经提交的任务完成, 包括尚未完成的任务 
        threadPool.shutdown();
        //会启动一个强制的关闭过程, 尝试取消所有运行中的任务和排在队列中尚未开始的任务，并把排队中尚未开始的任务返回，
        //对于关闭后提交到ExecutorService中的任务, 会被（拒接执行处理器）rejected execution handler处理，
        //它会抛弃任务，或者使得execute方法抛出一个未检查的RejectedExecutionException
        threadPool.shutdownNow();
    }

runable类：

static class PreRun2 implements Callable<Object>{
        private static int count;
        private final int id = count++;
        @Override
        public Object call() throws Exception {
            ThreadDemoUtils.sleep(1000);
            ThreadDemoUtils.printMessage("运行中");
            return id;
        }
    }

可以看到线程池的创建使用了

Executors.newFixedThreadPool(4);

我们来看下Executors，这个类主要是通过各中配置创建需要的线程池。

先来看看java线程池的类图关系

　　可以看到最顶层的接口是Executor 只提供了一个void execute(Runnable command)，这个方法是执行的意思，传进去一个参数然后执行，之后是ExecutorService接口，包含一些改变和观测线程池状态的方法，例如shutdownNow(),isShutdown(),还有一些执行线程的方法，例如submit()提交一个线程并且返回一个Future对象，Future可以获取线程运行的结果。AbstractExecutorService负责实现ExecutorService接口的一些方法。ThreadPoolExecutor继承了AbstractExecutorService重写了execute方法。

　　接下来我们看看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;
    }

我们来分析下这些参数

corePoolSize 

    核心线程池大小，这个参数标识线程池所可以容纳的最大的核心线程池数量，这个不是可以创建的最多的线程的数量，只是核心数量，我们之后会再次分析。

　maximumPoolSize

　线程池可以创建线程的最大数量。

keepAliveTime

　如果核心线程数已经满了，没有空闲的线程的时候，新的任务可以等待的最长时间。

workQueue

  当核心线程数量满了，没有空闲线程时，新任务放在这个指定的任务队列中。

threadFactory
 
创建线程的工厂类。

handler

 当无法创建更多的线程时拒绝任务的处理策略

我们看到上面使用线程池的例子中，通过

submit方法提交一个任务，然后得到一个Future对象，这个方法到底是怎么执行的呢，Future到底是什么？我们一起来看下

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

这个方法会先创建一个RunableFuture对象，就是将我们传进来的Runable封装成一个统一的任务对象

　　

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

     public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

看到会创建一个FutureTask对象，这个对象主要包含Callable和result ,如果传进来的是Runable,Executors.callable()方法会将runable转换为Callable对象（通过适配器模式）

也就是说不管传进来的是Runable还是Callable，他们都会将这个对象转换为FutreTask对象。

我们继续向下看

execute(ftask);

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();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

我们先来翻译一下这个注释

　　

 /*
         * 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.
　　有3个步骤：
　　　　1.如果当前创建的线程数量小于核心线程数量，就为传进来的任务创建一个新的线程，
　　　　　这个过程会调用addWorker方法原子的检查运行状态和工作线程来保证线程确实应该创建
　　　　　如果不应该创建就返回false，继续下一个步骤
　　　　2. 如果任务成功入队队列，任然需要检查我们应该添加一个线程或者这个线程池是不是已经挂了
　　　　3.如果我们不能入队，我们就尝试创建一个新的线程，如果创建失败，我们就会开启一个新的线程，
　　　　　　如果失败就会启动拒绝策略处理


         */

我们一个个方法来看，首先我们了解下线程池需要用到的所有状态

　　

/**
     * The main pool control state, ctl, is an atomic integer packing
     * two conceptual fields
     *   workerCount, indicating the effective number of threads
     *   runState,    indicating whether running, shutting down etc
     *
     * In order to pack them into one int, we limit workerCount to
     * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
     * billion) otherwise representable. If this is ever an issue in
     * the future, the variable can be changed to be an AtomicLong,
     * and the shift/mask constants below adjusted. But until the need
     * arises, this code is a bit faster and simpler using an int.
     *
     * The workerCount is the number of workers that have been
     * permitted to start and not permitted to stop.  The value may be
     * transiently different from the actual number of live threads,
     * for example when a ThreadFactory fails to create a thread when
     * asked, and when exiting threads are still performing
     * bookkeeping before terminating. The user-visible pool size is
     * reported as the current size of the workers set.
     *
     * The runState provides the main lifecycle control, taking on values:
     *
     *   RUNNING:  Accept new tasks and process queued tasks
     *   SHUTDOWN: Don\'t accept new tasks, but process queued tasks
     *   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
     *
     * The numerical order among these values matters, to allow
     * ordered comparisons. The runState monotonically increases over
     * time, but need not hit each state. The transitions are:
     *
     * RUNNING -> SHUTDOWN
     *    On invocation of shutdown(), perhaps implicitly in finalize()
     * (RUNNING or SHUTDOWN) -> STOP
     *    On invocation of shutdownNow()
     * SHUTDOWN -> TIDYING
     *    When both queue and pool are empty
     * STOP -> TIDYING
     *    When pool is empty
     * TIDYING -> TERMINATED
     *    When the terminated() hook method has completed
     *
     * Threads waiting in awaitTermination() will return when the
     * state reaches TERMINATED.
     *
     * Detecting the transition from SHUTDOWN to TIDYING is less
     * straightforward than you\'d like because the queue may become
     * empty after non-empty and vice versa during SHUTDOWN state, but
     * we can only terminate if, after seeing that it is empty, we see
     * that workerCount is 0 (which sometimes entails a recheck -- see
     * below).
     */

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY  = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;// 111 00000000000000000000000000000
    private static final int SHUTDOWN  =  0 << COUNT_BITS;// 000 00000000000000000000000000000
    private static final int STOP      =  1 << COUNT_BITS;// 001 00000000000000000000000000000
    private static final int TIDYING    =  2 << COUNT_BITS;// 010 00000000000000000000000000000
    private static final int TERMINATED =  3 << COUNT_BITS;// 100 00000000000000000000000000000

    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~CAPACITY; }//最高3位
    private static int workerCountOf(int c)  { return c & CAPACITY; }//后29位
    private static int ctlOf(int rs, int wc) { return rs | wc; }

以下是线程池状态的讲解

         RUNNING:  Accept new tasks and process queued tasks              运行时状态，可以接收新的任务
     *   SHUTDOWN: Don\'t accept new tasks, but process queued tasks       关闭状态，不在接收新的任务，会运行队列中的任务
     *   STOP:     Don\'t accept new tasks, don\'t process queued tasks,    立即关闭状态，不在接收新的任务，也不会执行队列中的任务，并且会中断运行中的任务
     *             and interrupt in-progress tasks
     *   TIDYING:  All tasks have terminated, workerCount is zero,        整理状态，所有的任务都结束了，工作任务为0，当STOP完成时。
     *             the thread transitioning to state TIDYING
     *             will run the terminated() hook method
     *   TERMINATED: terminated() has completed                           线程池完全结束

下面我们来看下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.
            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;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                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 {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

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

 这个方法整体分为两大部分，上面一部分是要将判断是不是可以添加一个线程，并且通过

compareAndIncrementWorkerCount(c)

来原子性的增加工作线程数量。下面一部分就是创建一个Worker工作线程，然后加锁将这个新建的worker放进workers中，并且启动这个工作线程，接下来我们看下Worker类

  private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;

        /**
         * 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
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

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

这个Worker实现了Runable接口，并持有一个Thread对象，当调用构造函数时这个Worker会创建一个线程并把Worker作为参数传进去

当该线程调用start方法时，就会调用系统api开辟一个新的线程，然后执行run方法

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

 看到runworker方法，他会不断的得到task，如果taks是null，就结束这个工作线程

processWorkerExit(w, completedAbruptly)

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
　　　　
        if (completedAbruptly) // If abrupt, then workerCount wasn\'t adjusted
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }

        tryTerminate();

        int c = ctl.get();
        if (runStateLessThan(c, STOP)) {
            if (!completedAbruptly) {
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

我们看到第一行，如果工作线程突然退出，就减少工作线程数量，然后将线程从workers中移除出去，然后开始尝试着结束线程池。

现在回到addWorker方法，这个方法是比较重要的方法，他负责创建并启动新的线程。能够保证线程数量不会超过限制。

继续看下execute方法

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

我们看到，如果线程数量小于核心线程数量，就继续创建线程，并将当前任务放置到创建的线程中运行，如果返回false说明线程池挂了或者线程数量超过了限制。

接下来判断线程池是否运行中，如果线程池运行中就执行入队列操作，将任务放在队列中，等待空闲的线程来执行该任务。我们接下来就看看队列的事

 

 

我们看到BlockingQueue有很多的实现，不同的实现对应不同的业务场景，我们来看下用的最多的LinkedBlockingQueue队列。

 public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

看下这个入队方法，如果队列的数量等于最大限制数量就返回false，入队失败，不然就加上一个入队的锁，然后再次判断是不是超过了最大限制数量，

如果没有超过就执行enqueue方法执行真正的入队方法。

    private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node;
    }

其实就是将节点追加到尾结点。

执行完入队操作后再次检查是否等于最大限制量，如果不等于就让非满条件进行通知。最后在finally中释放锁，方法最后检查是否入队成功并且只队列中只有一个，然后让非空条件进行通知。出队列也是一样的分析过程。

继续来看，入队之后会再次检查线程池的状态，是否可以继续运行，还会检查workerCountOf(c)==0,这个的目的是有可能在入队的时候其他的线程退出了导致没有线程可以使用，这个时候就要加入一个备用线程让进入队列的线程的运行得到保障。

最后判断addWorker(task,false)是否成功，如果成功就退出方法，如果失败就执行拒绝策略。