JDK自带线程池学习

JDK自带线程池

线程池的状态

线程有如下状态

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.

    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; // -1为全1，所以我们左移29位就是111开头的状态位
    private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000
    private static final int STOP       =  1 << COUNT_BITS; // 001
    private static final int TIDYING    =  2 << COUNT_BITS; // 010
    private static final int TERMINATED =  3 << COUNT_BITS; // 011

    // Packing and unpacking ctl
    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; }

我们从上面的源代码中可以看出，我们将状态存储在一个原子整型的前三位，然后将线程的容量存储在后29位。将状态和容量放在一起，这样更新状态和容量只需要进行一次cas操作。

下面的三个方法就是进行获取状态和工作线程数量和初始化状态。

在源代码注释中解释到，当未来原子整型不够用了，就会将其升级为原子长整形。且状态位也有扩展的空间，如果需要的话。

同时源代码中也有表明各个状态转换的条件，可以ThreadPoolExecutor类中下载source查看。

线程池构造方法

参数组成

corePoolSize核心线程的数量
maximumPoolSize最大的线程数量 PS：最大线程数-核心线程数 = 急救线程的数量
keepAliveTime 急救线程的存活时间
unit 急救线程存活时间单位
workQueue 阻塞队列
threadFactory 线程工厂 PS：线程工厂就是创造线程的工厂，为其进行给任务和名字
handler 拒绝策略的实现 PS：就是当阻塞队列满了之后所要做的动作，死等，限时等(RocketMQ)，交给调用者运行，直接抛弃，创建一个新线程(netty)，抛出异常写日志（dubbo）。

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

这就是拒绝策略在JDK自带的实现。

AbortPolicy直接抛出一个RejectedExecutionException异常，dubbo应该是加以记录一些更多的信息猜测
CallerRunsPolicy就是让调用者自己执行这个任务
DiscardPolicy直接抛弃
DiscardOldestPolicy抛弃早进入阻塞队列的然后让当前任务进入阻塞队列

JDK线程池和上一次的线程池不一样的就是急救线程。

急救线程就是当阻塞队列满了之后，并不会像我上次的例子一样直接进行拒绝策略的判断，会创建一个急救线程或者已存活的急救线程进行执行任务，如果急救线程也满了的话，才会进入拒绝策略的判断。

JDK线程池的基本使用

固定大小线程池

ExecutorService executorService = Executors.newFixedThreadPool(2, new ThreadFactory() {
    AtomicInteger ctl = new AtomicInteger();

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r,"myThreadPoll-" + ctl.getAndIncrement());
        return thread;
    }
});
executorService.execute(()->{
    log.debug("线程执行了一次");
});
executorService.execute(()->{
    log.debug("线程执行了一次");
});
executorService.execute(()->{
    log.debug("线程执行了一次");
});

我看了一下它的默认构造方法，它创造了一个Integer.MAX_VALUE大小的阻塞队列。阻塞队列其实和小测验的阻塞队列是差不多的。

缓存线程池

主要是它的阻塞队列的不同，其中核心数为0，然后通过阻塞队列直到有线程对其进行取任务，不然就是一直阻塞的状态。

@Slf4j
public class Test2 {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(()->{
            log.debug("执行成功");
        });
        executorService.execute(()->{
            log.debug("执行成功");
        });
    }

}

A blocking queue in which each insert operation must wait for a corresponding remove operation by another thread, and vice versa.

翻译：一种阻塞队列，其中每个插入操作必须等待另一个线程执行相应的删除操作，反之亦然。

单线程线程池

顾名思义即单线程的线程池，不过有意思的一点就是这个使用了一个设计模式就是装饰器模式。

@Slf4j
public class Test3 {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(()->{
            log.debug("hello");
        });
        executorService.execute(()->{
            log.debug("hello");
        });
    }

}

首先是因为如果我们直接返回ThreadPoolExecutor这个类的话，我们是知道了它的类，是可以直接使用强转来实现修改线程的核心数以及一些参数。如下

@Slf4j
public class Test1 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2, new ThreadFactory() {
            AtomicInteger ctl = new AtomicInteger();

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r,"myThreadPoll-" + ctl.getAndIncrement());
                return thread;
            }
        });
		// 我们将其对象通过强转直接修改了其的核心数，执行结构同样生效
        ThreadPoolExecutor executor= (ThreadPoolExecutor) executorService;
        executor.setCorePoolSize(1);
        executorService.execute(()->{
            log.debug("线程执行了一次");
        });
        executorService.execute(()->{
            log.debug("线程执行了一次");
        });
        executorService.execute(()->{
            log.debug("线程执行了一次");
        });
    }
}

但是如果我们通过装饰器模式将其进行包装，然后包装的对象返回，是无法进行修改核心数的，更何况单线程线程池的情况下，我们需要保证核心数总为1把，不能让其他人修改。展示部分代码，返回的是其的装饰类。

static class DelegatedExecutorService extends AbstractExecutorService {
    private final ExecutorService e;
    DelegatedExecutorService(ExecutorService executor) { e = executor; }
    public void execute(Runnable command) { e.execute(command); }
    public void shutdown() { e.shutdown(); }
    public List<Runnable> shutdownNow() { return e.shutdownNow(); }

ExecutorService方法使用

这个类就是线程池的接口类，掌管着线程池的方法。

void shutdown() 继续执行当前线程池中的任务和阻塞队列中的任务，不再接收新的任务。
List shutdownNow() 尝试停止所有正在执行的任务，停止正在等待的任务的处理，并返回正在等待执行的任务的列表
boolean isShutdown() 返回当前线程池是否已经关闭
boolean isTerminated() 在调用了shutdown或Shutdownow后，关闭后所有任务都已完成，则返回true。如果没有调用永远不会返回true。
boolean awaitTermination(long timeout, TimeUnit unit) 阻塞，直到所有任务在关闭请求后完成执行，或超时发生，或当前线程中断，以先发生的为准。
Future submit(Callable task) 返回执行结果
List<Future> invokeAll(Collection<? extends Callable> tasks) 批量返回结果
T invokeAny(Collection<? extends Callable> tasks) 执行任意一个并返回结果，如果任何一个完成，其他正在执行的直接结束。

工作线程的饥饿现象

/**
 * @Author 10276
 * @Date 2022/5/11 20:52
 */
@Slf4j
public class Test4 {

    public static void main(String[] args) {
        ExecutorService rest = Executors.newFixedThreadPool(2);
        rest.execute(()->{
            log.debug("准备点餐");
            Future<String> submit = rest.submit(() -> {
                log.debug("正在坐菜");
                return "菜";
            });
            try {
                log.debug("上菜{}",submit.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
        rest.execute(()->{
            log.debug("准备点餐");
            Future<String> submit = rest.submit(() -> {
                log.debug("正在坐菜");
                return "菜";
            });
            try {
                log.debug("上菜{}",submit.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}

一个线程池的导致两个线程没办法继续进行下去，没有多余的线程取做接下来的事情，但这不是死锁问题，归结原因还是线程资源不够，同时也无法继续进行下去了。

解决方案：由此可以得出对于线程池数量的选择和线程池中核心线程数量的选择是十分重要的。

public static void main(String[] args) {
    ExecutorService waitress = Executors.newFixedThreadPool(1);
    ExecutorService cooker = Executors.newFixedThreadPool(1);
    waitress.execute(()->{
        log.debug("准备点餐");
        Future<String> submit = cooker.submit(() -> {
            log.debug("正在做菜");
            return "湖南菜";
        });
        try {
            log.debug("上菜:{}",submit.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    });
    waitress.execute(()->{
        log.debug("准备点餐");
        Future<String> submit = cooker.submit(() -> {
            log.debug("正在做菜");
            return "广东菜";
        });
        try {
            log.debug("上菜：{}",submit.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    });
}

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/14534.html