线程池原理及使用

2024-04-11 23:17:37

1.线程池核心参数含义

线程池源码，咱们先依次介绍下各个参数的含义：

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

corePoolSize：核心线程数

maximumPoolSize：最大线程数

keepAliveTime：当线程数大于核心数时，这是多余的空闲线程在终止前等待新任务的最长时间；即空闲线程没任务执行时存活的最长时间

unit：时间单位，针对第三个参数keepAliveTime

workQueue：线程池中存放任务的阻塞队列，有界阻塞队列/*阻塞队列

threadFactory：执行程序创建新线程时使用的工厂

handler：由于达到线程边界和队列容量而阻塞执行时要使用的处理程序；jdk中自带了四种处理策略：

1.AbortPolicy:线程池默认策略，当线程池队列满时，再有任务过来，直接丢弃任务，并抛出RejectedExecutionException异常

//JDK 1.8源码

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

2.CallerRunsPolicy:被拒绝任务的处理程序，它直接在主线程中运行被拒绝的任务，除非主线程被关闭，在这种情况下任务将被丢弃。意思就是任务添加到队列失败，此时主线程直接执行此任务，除非主线程是被杀了才不会执行

//JDK1.8源码

public static class CallerRunsPolicy implements RejectedExecutionHandler { 
    public CallerRunsPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }

3.DiscardOldestPolicy：抛弃队列中最老的任务，然后将新任务尝试入队，队列是先进先出，所以队头的任务肯定是最老的

//JDK1.8源码

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

4.DiscardPolicy：队列满的话丢掉新任务但是不会抛出异常

//JDK1.8源码

public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        //空方法
    }
}

这个策略大家可以根据自己业务需求设置，也可以自定义写自己的异常策略。

2.底层原理

AtomicInteger atomicInteger = new AtomicInteger(0);
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 
2, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(1),
        (r) -> new Thread(r, "t" + atomicInteger.incrementAndGet()),
        new ThreadPoolExecutor.AbortPolicy());

上面是我自己定义的一个线程池：核心线程数1，最大线程数2，空闲线程超时时间60秒，定义的长度为1的有界队列，拒绝策略采用AbortPolicy策略

假如现在同时过来4个任务：task1由核心线程执行，task2会进入队列，task3和task4发现核心线程和队列都满，则会新建空闲线程(1个空闲线程)，task3或者task4其中的一个会被空闲线程执行，剩余的一个则执行策略，抛出报错

任务进入线程池的底层逻辑：判断核心线程数是否还有剩余，若有剩余，任务直接由核心线程执行；若达到核心线程数，则会进入队列，队列分有界阻塞队列和*阻塞队列，若是*队列，按理说可以入队无限多的任务(*队列可能会引起内存相关问题，下文讲解)；若是有界队列，则判断是否达到有界队列的最大长度，若队列没满，则任务入队；若队列已满，则会执行相应的策略。

核心线程是不会被销毁的，任务执行完毕后，会从队列中拿出队头的任务接着执行

空闲线程在超过设置的超时时间未执行任务则会被销毁

3.线程池在使用过程中的问题

3.1 如果在线程池中使用*阻塞队列会有什么问题？

若队列里面的任务在访问远程服务异常的情况下会调用失败导致超时，导致队列里面积压的任务越来越多，执行任务的线程在访问远程服务都会超时，时间很漫长，但是任务入队列的速度还是一样的快，导致*阻塞队列越来越大，会导致内存异常的飙升，可能还会导致OOM；

3.2 若线程池的队列满了之后，会发生什么?

若你new的线程池的阻塞队列是有界的，可以避免内存溢出，在队列满之后，此时需要看你设置的maximumPoolSize值，若这个值为（1000）则可能存在待执行的任务瞬间超过这个值，用自带的策略的话，超过的任务会被抛弃，导致任务丢失；若maximumPoolSize=Integer.MAX_VALUE,多少任务就会创建多少线程，但是每个线程会有自己的栈内存，占用一定的内存空间，这样也可能会导致内存飙升，系统会崩溃掉，即使内存没有溢出，也会导致机器的CPU负载特别高导致机器挂掉；

若用*队列，则会出现上面3.1的问题

具体的如何设计线程池需要结合自己业务场景、业务负载制定合适的线程池参数；或者自定义策略，若线程池已经无法执行更多的任务，可以把任务持久化写入磁盘，在后台专门起一个线程，等线程池的工作负载降低了，再执行磁盘里的任务等

3.3 若服务宕机了，阻塞队列里面的请求任务该如何处理？

服务器宕机，队列存储在内存中，则队列里面的任务都会丢失，为了解决这一问题，我们可以在提交任务到线程池之前将任务插入数据库，记录此任务的状态(未提交等)，在服务器重启后，后台设计一个功能在此数据库中将漏掉的任务再次提交即可；

学无止尽，与君共勉！！！

码农公寓

相关文章