高并发之——ThreadPoolExecutor类居然是这样保证线程池正确运行的...

简介: 对于线程池的核心类ThreadPoolExecutor来说,有哪些重要的属性和内部类为线程池的正确运行提供重要的保障呢?

问题

对于线程池的核心类ThreadPoolExecutor来说,有哪些重要的属性和内部类为线程池的正确运行提供重要的保障呢?

ThreadPoolExecutor类中的重要属性

在ThreadPoolExecutor类中,存在几个非常重要的属性和方法,接下来,我们就介绍下这些重要的属性和方法。

ctl相关的属性

AtomicInteger类型的常量ctl是贯穿线程池整个生命周期的重要属性,它是一个原子类对象,主要用来保存线程的数量和线程池的状态,我们看下与这个属性相关的代码如下所示。

//主要用来保存线程数量和线程池的状态,高3位保存线程状态,低29位保存线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程池中线程的数量的位数(32-3)
private static final int COUNT_BITS = Integer.SIZE - 3;
//表示线程池中的最大线程数量
//将数字1的二进制值向右移29位,再减去1
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
//线程池的运行状态
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;
//获取线程状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获取线程数量
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}
private boolean compareAndIncrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect + 1);
}
private boolean compareAndDecrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect - 1);
}
private void decrementWorkerCount() {
    do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}

对于线程池的各状态说明如下所示。

  • RUNNING:运行状态,能接收新提交的任务,并且也能处理阻塞队列中的任务
  • SHUTDOWN: 关闭状态,不能再接收新提交的任务,但是可以处理阻塞队列中已经保存的任务,当线程池处于RUNNING状态时,调用shutdown()方法会使线程池进入该状态
  • STOP: 不能接收新任务,也不能处理阻塞队列中已经保存的任务,会中断正在处理任务的线程,如果线程池处于RUNNING或SHUTDOWN状态,调用shutdownNow()方法,会使线程池进入该状态
  • TIDYING: 如果所有的任务都已经终止,有效线程数为0(阻塞队列为空,线程池中的工作线程数量为0),线程池就会进入该状态。
  • TERMINATED: 处于TIDYING状态的线程池调用terminated ()方法,会使用线程池进入该状态

也可以按照ThreadPoolExecutor类的注释,将线程池的各状态之间的转化总结成如下图所示。

微信图片_20211118115640.jpg

  • RUNNING -> SHUTDOWN:显式调用shutdown()方法, 或者隐式调用了finalize()方法
  • (RUNNING or SHUTDOWN) -> STOP:显式调用shutdownNow()方法
  • SHUTDOWN -> TIDYING:当线程池和任务队列都为空的时候
  • STOP -> TIDYING:当线程池为空的时候
  • TIDYING -> TERMINATED:当 terminated() hook 方法执行完成时候

其他重要属性

除了ctl相关的属性外,ThreadPoolExecutor类中其他一些重要的属性如下所示。

//用于存放任务的阻塞队列  
private final BlockingQueue<Runnable> workQueue;
//可重入锁
private final ReentrantLock mainLock = new ReentrantLock();
//存放线程池中线程的集合,访问这个集合时,必须获得mainLock锁
private final HashSet<Worker> workers = new HashSet<Worker>();
//在锁内部阻塞等待条件完成
private final Condition termination = mainLock.newCondition();
//线程工厂,以此来创建新线程
private volatile ThreadFactory threadFactory;
//拒绝策略
private volatile RejectedExecutionHandler handler;
//默认的拒绝策略
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

ThreadPoolExecutor类中的重要内部类

在ThreadPoolExecutor类中存在对于线程池的执行至关重要的内部类,Worker内部类和拒绝策略内部类。接下来,我们分别看这些内部类。

Worker内部类

Worker类从源代码上来看,实现了Runnable接口,说明其本质上是一个用来执行任务的线程,接下来,我们看下Worker类的源代码,如下所示。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
    private static final long serialVersionUID = 6138294804551838833L;
    //真正执行任务的线程
    final Thread thread;
    //第一个Runnable任务,如果在创建线程时指定了需要执行的第一个任务
    //则第一个任务会存放在此变量中,此变量也可以为null
    //如果为null,则线程启动后,通过getTask方法到BlockingQueue队列中获取任务
    Runnable firstTask;
    //用于存放此线程完全的任务数,注意:使用了volatile关键字
    volatile long completedTasks;
    //Worker类唯一的构造放大,传递的firstTask可以为null
    Worker(Runnable firstTask) {
        //防止在调用runWorker之前被中断
        setState(-1);
        this.firstTask = firstTask;
        //使用ThreadFactory 来创建一个新的执行任务的线程
        this.thread = getThreadFactory().newThread(this);
    }
    //调用外部ThreadPoolExecutor类的runWorker方法执行任务
    public void run() {
        runWorker(this);
    }
    //是否获取到锁 
    //state=0表示锁未被获取
    //state=1表示锁被获取
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }
    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

在Worker类的构造方法中,可以看出,首先将同步状态state设置为-1,设置为-1是为了防止runWorker方法运行之前被中断。这是因为如果其他线程调用线程池的shutdownNow()方法时,如果Worker类中的state状态的值大于0,则会中断线程,如果state状态的值为-1,则不会中断线程。

Worker类实现了Runnable接口,需要重写run方法,而Worker的run方法本质上调用的是ThreadPoolExecutor类的runWorker方法,在runWorker方法中,会首先调用unlock方法,该方法会将state置为0,所以这个时候调用shutDownNow方法就会中断当前线程,而这个时候已经进入了runWork方法,就不会在还没有执行runWorker方法的时候就中断线程。

注意:大家需要重点理解Worker类的实现。

拒绝策略内部类

在线程池中,如果workQueue阻塞队列满了,并且没有空闲的线程池,此时,继续提交任务,需要采取一种策略来处理这个任务。而线程池总共提供了四种策略,如下所示。

  • 直接抛出异常,这也是默认的策略。实现类为AbortPolicy。
  • 用调用者所在的线程来执行任务。实现类为CallerRunsPolicy。
  • 丢弃队列中最靠前的任务并执行当前任务。实现类为DiscardOldestPolicy。
  • 直接丢弃当前任务。实现类为DiscardPolicy。

在ThreadPoolExecutor类中提供了4个内部类来默认实现对应的策略,如下所示。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}
public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
    }
}
public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

我们也可以通过实现RejectedExecutionHandler接口,并重写RejectedExecutionHandler接口的rejectedExecution方法来自定义拒绝策略,在创建线程池时,调用ThreadPoolExecutor的构造方法,传入我们自己写的拒绝策略。

例如,自定义的拒绝策略如下所示。

public class CustomPolicy implements RejectedExecutionHandler {
    public CustomPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            System.out.println("使用调用者所在的线程来执行任务")
            r.run();
        }
    }
}

使用自定义拒绝策略创建线程池。

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                       60L, TimeUnit.SECONDS,
                       new SynchronousQueue<Runnable>(),
                       Executors.defaultThreadFactory(),
               new CustomPolicy());
相关文章
|
1月前
|
安全 Java
Java多线程集合类
本文介绍了Java中线程安全的问题及解决方案。通过示例代码展示了使用`CopyOnWriteArrayList`、`CopyOnWriteArraySet`和`ConcurrentHashMap`来解决多线程环境下集合操作的线程安全问题。这些类通过不同的机制确保了线程安全,提高了并发性能。
|
2月前
lua面向对象(类)和lua协同线程与协同函数、Lua文件I/O
Lua的面向对象编程、协同线程与协同函数的概念和使用,以及Lua文件I/O操作的基本方法。
37 4
lua面向对象(类)和lua协同线程与协同函数、Lua文件I/O
|
2月前
|
Java 开发者
在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口
【10月更文挑战第20天】在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口。本文揭示了这两种方式的微妙差异和潜在陷阱,帮助你更好地理解和选择适合项目需求的线程创建方式。
27 3
|
2月前
|
Java
在Java多线程编程中,实现Runnable接口通常优于继承Thread类
【10月更文挑战第20天】在Java多线程编程中,实现Runnable接口通常优于继承Thread类。原因包括:1) Java只支持单继承,实现接口不受此限制;2) Runnable接口便于代码复用和线程池管理;3) 分离任务与线程,提高灵活性。因此,实现Runnable接口是更佳选择。
51 2
|
2月前
|
Java
Java中多线程编程的基本概念和创建线程的两种主要方式:继承Thread类和实现Runnable接口
【10月更文挑战第20天】《JAVA多线程深度解析:线程的创建之路》介绍了Java中多线程编程的基本概念和创建线程的两种主要方式:继承Thread类和实现Runnable接口。文章详细讲解了每种方式的实现方法、优缺点及适用场景,帮助读者更好地理解和掌握多线程编程技术,为复杂任务的高效处理奠定基础。
38 2
|
2月前
|
Java 开发者
Java多线程初学者指南:介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点
【10月更文挑战第20天】Java多线程初学者指南:介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点,重点解析为何实现Runnable接口更具灵活性、资源共享及易于管理的优势。
44 1
|
2月前
|
Java Linux
【网络】高并发场景处理:线程池和IO多路复用
【网络】高并发场景处理:线程池和IO多路复用
58 2
|
2月前
|
Java C++
【多线程】JUC的常见类,Callable接口,ReentranLock,Semaphore,CountDownLatch
【多线程】JUC的常见类,Callable接口,ReentranLock,Semaphore,CountDownLatch
36 0
|
2月前
|
存储 消息中间件 资源调度
C++ 多线程之初识多线程
这篇文章介绍了C++多线程的基本概念,包括进程和线程的定义、并发的实现方式,以及如何在C++中创建和管理线程,包括使用`std::thread`库、线程的join和detach方法,并通过示例代码展示了如何创建和使用多线程。
58 1
C++ 多线程之初识多线程
|
2月前
|
Java 开发者
在Java多线程编程中,选择合适的线程创建方法至关重要
【10月更文挑战第20天】在Java多线程编程中,选择合适的线程创建方法至关重要。本文通过案例分析,探讨了继承Thread类和实现Runnable接口两种方法的优缺点及适用场景,帮助开发者做出明智的选择。
23 2
下一篇
DataWorks