线程池中的2个注意点

线程池在日常开发中多多少少都会接触和使用.
其中和线程池关系最为紧密的一个就是阻塞队列,用于存储提交到线程池中的任务.
关于向阻塞队列中添加任务和获取任务会涉及到很多方法,如下

image.png

那么当我们向线程池提交任务的时候,它会调用上面的哪个方法呢?

// 代码位置: java.util.concurrent.ThreadPoolExecutor#execute

public void execute(Runnable command) {
    
    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);
}

从源码中我们知道,在向线程池阻塞队列中提交任务时,调用的是offer(command)方法.
也就是说,它并不会阻塞提交任务的线程.

在线程池中的线程会不停的从阻塞队列中获取任务,那么它们又是调用的哪个方法呢?

// 代码位置: java.util.concurrent.ThreadPoolExecutor#getTask
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

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

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // 超时获取或阻塞获取任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

从源码中我们知道,线程池中的线程在向阻塞队列获取任务时,通过超时获取或者阻塞获取的方式.

image.png

讨论第二个注意点
我们在学习ReentrantLock的时候,手动加锁和释放锁必须是成对出现的,这也是我们大家一贯的认知.

// 代码位置: java.util.concurrent.ThreadPoolExecutor#runWorker

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

代码中,我们可以看到lock()和unlokc()成对出现了.可是奇怪的是,在方法开始处还有一个unlock()调用.

w.unlock(); // allow interrupts

在没有出现lock()的情况下,居然出现了unlock()调用.作者还注释了allow interrupts.
是不是觉得有点和之前自己的认知有点反差.

这里我们简单介绍下线程池中的Worker这个类.

// 代码位置: java.util.concurrent.ThreadPoolExecutor.Worker

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {

}

它继承了AQS,自己实现了加锁和释放锁,并没有使用ReentrantLock.
它并没有像ReentrantLock那样是可重入的,它不允许重入.
而且在它的构造函数中,直接将state设置成-1,并没有使用默认值0.

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

在构造函数中,将state默认值本应该为0,被设置成了-1,相当于lock()的语义.
而且源码注释上也说明了,禁止中断直到运行runWorker. 因此我们才会看到上面那个一开始就调用了unlock()方法.
关于基于AQS实现各种锁,以及操作state这个值,我们必须灵活使用.不能被常规认识所蒙蔽.