Java 线程 — ThreadPoolExecutor

线程池

线程池处理流程

  • 核心线程池:创建新线程执行任务,需要获取全局锁
  • 队列:将新来的任务加入队列
  • 线程池:大于corePoolSize,并且队列已满,小于maxPoolSize,创建新的worker执行任务
  • 线程池已满(达到max)处理策略:大于线程最大处理能力,大于maxPoolSize,选择拒绝策略

尽可能避免获取全局锁,corePoolSize就是这个作用,线程池开始处理任务,预热达到corePoolSize之后,将新来的任务放入队列

execute

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // ctl是一个AtomInteger,低29为表示线程数workerCount,高3位表示线程池运行状态runState
    int c = ctl.get();
    // 当前前程数小于核心线程数corePoolSize,新建核心线程处理任务
    if (workerCountOf(c) < corePoolSize) {
        // 第二个参数为true表示新建的核心线程
        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);
    }
    // 如果放入队列失败(队列已满),则新建线程处理任务,第二个参数为false表示新建的是非核心线程
    else if (!addWorker(command, false))
        // 如果已经超过最大线程数,采取相应的拒绝策略
        reject(command);
}

// 第一个参数是需要运行的任务
// 第二个参数是为了区分核心线程和非核心线程,用来确定线程池的边界是corePoolSize还是maxPoolSize
// 本方法的作用就是新建一个worker线程并启动
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;

        // 循环确保CAS操作成功
        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 成功增加workerCount则跳出外层循环,开始新建线程
            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 {
        final ReentrantLock mainLock = this.mainLock;
        // 新建worker线程
        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();
                    // 将新建的线程放入线程池中,其实就是一个HashSet
                    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;
}

创建线程

// 在addWorker中新建线程
w = new Worker(firstTask);

// Worker的构造方法
Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

// 这里以ThreadFactory的一个实现DefaultThreadFactory为例
DefaultThreadFactory() {
    SecurityManager s = System.getSecurityManager();
    // 新建一个线程组,线程池中所有线程都由线程组统一管理
    group = (s != null) ? s.getThreadGroup() :
                          Thread.currentThread().getThreadGroup();
    namePrefix = "pool-" +
                  poolNumber.getAndIncrement() +
                 "-thread-";
}

// 新建线程
public Thread newThread(Runnable r) {
    // 新建的线程属于线程组group,r为Worker对象
    Thread t = new Thread(group, r,
                          namePrefix + threadNumber.getAndIncrement(),
                          0);
    if (t.isDaemon())
        t.setDaemon(false);
    if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
    return t;
}

线程池中的worker线程执行任务

// 在addWorker中会启动worker线程
if (workerAdded) {
    t.start();
    workerStarted = true;
}

// 线程启动之后会执行worker的run方法
// 调用外部类ThreadPoolExecutor的runWorker方法
public void run() {
    runWorker(this);
}

// runWorker,worker线程循环执行来自workerQueue的任务(除firstTask外)
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 新建线程以后会执行第一个任务,新建Worker线程的时候传入的任务
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // worker不断从workerQueue中getTask执行,
        // 如果没有获取任务并且返回,则while循环结束,worker线程结束
        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 {
                    // 调用run方法执行
                    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);
    }
}

// 从workerQueue中获取task,workerQueue是BlockingQueue
// 如果返回null,则会导致worker线程结束
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);
            // 当allowCoreThreadTimeOut为true或者当前线程数大于核心线程数的时候timed为true
            // allowCoreThreadTimeOut为true:允许线程没有任务的时候超时退出
            // wc > corePoolSize:表示当前线程数足够,可以结束,以维持核心线程数
            timed = allowCoreThreadTimeOut || wc > corePoolSize;

            if (wc <= maximumPoolSize && ! (timedOut && timed))
                break;
            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 {
            // workQueue.poll:等待keepAliveTime纳秒之后,如果获取到则返回任务,否则返回null
            // workQueue.take:如果没有获取到任务,线程会一直阻塞,直到被唤醒(什么时候会被唤醒:新加入task的时候,poll后队列中元素数依然大于1个,队列中元素数等于capacity),所以take如果返回的话一定是非空的
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

线程池满之后的拒绝策略

CallerRunsPolicy

直接在调用线程池所在的线程运行(调用)

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        // 直接调用run方法
        r.run();
    }
}

AbortPolicy

throw Exception

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

DiscardPolicy

直接丢弃,不做任何操作

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}

DiscardOldestPolicy

丢弃最老任务的策略

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        // 移除第一个任务,也就是最老(最先进来的)任务
        e.getQueue().poll();
        // 执行当前传入的任务
        e.execute(r);
    }
}

问题

线程池的“池”体现在哪里

  • 平时使用线程都是直接继承Thread然后,调用start,然后Thread.start通过native调用创建新的线程并回调自己定义的run方法,所以每次调用一次run方法都需要新建一个线程
  • 线程池就是,启动一个线程(Worker)调用完一个run(直接调用,不再通过Thread.start)方法之后并不会立即退出,会运行在队列中等待的其他run方法(这里是循环,从队列中获取),如果队列中没有需要继续运行的线程则当前worker线程会休眠
    原文作者:lacker
    原文地址: http://www.cnblogs.com/sunshine-2015/p/6080174.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞