网站首页 > 技术文章正文

学习线程池掌握治理线程的法宝讲讲线程池的实现原理

btikc 2024-10-22 10:30:51 技术文章 8 ℃ 0 评论

推荐阅读：

1.为什么需要线程池

在当今计算机的CPU计算速度非常快的情况下，为了能够充分利用CPU性能提高程序运行效率我们在程序中使用了线程。但是在高并发情况下会频繁的创建和销毁线程，这样就变相的阻碍了程序的执行速度，所以为了管理线程资源和减少线程创建以及销毁的性能消耗就引入了线程池。

2.什么场景下适合使用线程池

当服务器接收到大量任务时，如果使用线程池可以大量减少线程的创建与销毁次数，从而提升程序执行效率
在实际开发中，如果需要创建5个以上的线程，那么就可以使用线程池来管理

3.线程池参数介绍以及特点

3.1 corePoolSize和maxPoolSize

corePoolSize：线程池在创建完时，里面并没有线程，只有当任务到来时再去创建线程。

maxPoolSize：线程池可能会在核心线程数的基础上额外增加一些线程，但是线程数量的上限是maxPoolSize。比如第一天执行的任务非常多，第二天执行的任务非常少，但是有了maxPoolSize参数，就可以增强任务处理的灵活性。

3.2 添加线程的规则

当线程数量小于corePoolSize即使线程没有在执行任务，也会创建新的线程。
如果线程数量等于（或大于）corePoolSize，但小于maxPoolSize则将任务放入队列。
如果队列已满，并且线程数小于maxPoolSize，则创建新的线程运行任务。
如果队列已满，并且线程数大于或等于maxPoolSize，则拒绝该任务。

执行流程：

3.3 增减线程的特点

将corePoolSize和maxPoolSize设置为相同的值，那么就会创建固定大小的线程池。
线程池希望保持更少的线程数，并且只有在负载变得很大时才会增加它。
如果将线程池的maxPoolSize参数设置为很大的值，例如Integer.MAX_VALUE，可以允许线程池容纳任意数量的并发任务。
只有在队列满了的时候才会去创建大于corePoolSize的线程，所以如果使用了无界队列（如：LinkedBlockingQueue）就不会创建到超过corePoolSize的线程数。

3.4 keepAliveTime

如果线程池当前的线程数大于corePoolSize,那么如果多余的线程的空闲时间大于keepAliveTime，它们就会被终止。

keepAliveTime参数的使用可以减少线程数过多冗余时的资源消耗。

3.5 threadFactory

新的线程由ThreadFactory创建，默认使用Executors.defaultThreadFactory()，创建出来的线程都在同一个线程组，拥有同样的NORM_PRIORITY优先级并且都不是守护线程。如果自己指定ThreadFactory，那么就可以改变线程名、线程组、优先级、是否是守护线程等。通常情况下直接使用defaultThreadFactory就行。

3.6 workQueue

直接交接（SynchronousQueue）：任务不多时，只需要用队列进行简单的任务中转，这种队列无法存储任务，在使用这种队列时，需要将maxPoolSize设置的大一点。
无界队列（LinkedBlockingQueue）：如果使用无界队列当作workQueue，将maxQueue设置的多大都没有用，使用无界队列的优点是可以防止流量突增，缺点是如果处理任务的速度跟不上提交任务的速度，这样就会导致无界队列中的任务越来越多，从而导致OOM异常。
有界队列（ArrayBlockingQueue）：使用有界队列可以设置队列大小，让线程池的maxPoolSize有意义。

4.线程池应该手动创建还是自动创建

手动创建更好，因为这样可以让我们更加了解线程池的运行规则，避免资源耗尽的风险。

4.1 直接调用JDK封装好的线程池会带来的问题

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newFixedThreadPool线程池通过传入相同的corePoolSize和maxPoolSize可以保证线程数量固定，0L的keepAliveTime表示时刻被销毁，workQueue使用的是无界队列。这样潜在的问题就是当处理任务的速度赶不上任务提交的速度的时候，就可能会让大量任务堆积在workQueue中，从而引发OOM异常。

4.2 演示newFixedThreadPool内存溢出问题

/**
 * 演示newFixedThreadPool线程池OOM问题
 */
public class FixedThreadPoolOOM {

    private static ExecutorService executorService = Executors.newFixedThreadPool(1);

    public static void main(String[] args) {
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            executorService.execute(new SubThread());
        }
    }
}

class SubThread implements Runnable {

    @Override
    public void run() {
        try {
            //延长任务时间
            Thread.sleep(1000000000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

更改JVM参数

运行结果

4.3 newSingleThreadExecutor

使用线程池打印线程名

public class SingleThreadExecutor {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 1000; i++) {
            executorService.execute(new Task());
        }
    }
}

查看newSingleThreadExecutor源码

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

从源码可以看出newSingleThreadExecutor和newFixedThreadPool基本类似，不同的只是corePoolSize和maxPoolSize的值，所以newSingleThreadExecutor也存在内存溢出问题。

4.4 newCachedThreadPool

newCachedThreadPool也被称为可缓存线程池，它是一个无界线程池，具有自动回收多余线程的功能。

public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 1000; i++) {
            executorService.execute(new Task());
        }
    }
}

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

newCachedThreadPool的maxPoolSize设置的值为Integer.MAX_VALUE，所以可能会导致线程被无限创建，最终导致OOM异常。

4.5 newScheduledThreadPool

该线程池支持周期性任务的执行

public class ScheduledThreadPoolTest {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledExecutorService =
                Executors.newScheduledThreadPool(10);
//        scheduledExecutorService.schedule(new Task(), 5, TimeUnit.SECONDS);
        scheduledExecutorService.scheduleAtFixedRate(new Task(), 1, 3, TimeUnit.SECONDS);
    }
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

4.6 正确的创建线程池的方法

根据业务场景不同，自己设置线程池参数，例如内存有多大，自己取线程名子等。

4.7 线程池里的线程数量设置多少比较合适？

CPU密集型（加密、计算hash等）：最佳线程数设置为CPU核心数的1——2倍。
耗时I/O型（读写数据库、文件、网络读写等）：最佳线程数一般会大于CPU核心数很多倍，以JVM监控显示繁忙情况为依据，保证线程空闲可以衔接上。参考Brain Goezt推荐的计算方法：

线程数=CPU核心数 × (1+平均等待时间/平均工作时间)

5.对比线程池的特点

FixedThreadPool：通过手动传入corePoolSize和maxPoolSize，以固定的线程数来执行任务
SingleThreadExecutor：corePoolSize和maxPoolSize默认都是1，全程只以1条线程执行任务
CachedThreadPool：它没有需要维护的核心线程数，每当需要线程的时候就进行创建，因为它的线程存活时间是60秒，所以它也凭借着这个参数实现了自动回收的功能。
ScheduledThreadPool：这个线程池可以执行定时任务，corePoolSize是通过手动传入的，它的maxPoolSize为Integer.MAX_VALUE，并且具有自动回收线程的功能。

5.1 为什么FixedThreadPool和SingleThreadExecutor的Queue是LinkedBlockingQueue？

因为这两个线程池的核心线程数和最大线程数都是相同的，也就无法预估任务量，所以需要在自身进行改进，就使用了无界队列。

5.2 为什么CachedThreadPool使用的Queue是SynchronousQueue？

因为缓存线程池的最大线程数是“无上限”的，每当任务来的时候直接创建线程进行执行就好了，所以不需要使用队列来存储任务。这样避免了使用队列进行任务的中转，提高了执行效率。

5.3 为什么ScheduledThreadPool使用延迟队列DelayedWorkQueue？

因为ScheduledThreadPool是延迟任务线程池，所以使用延迟队列有利于对执行任务的时间做延迟。

5.4 JDK1.8中加入的workStealingPool

workStealingPool适用于执行产生子任务的环境，例如进行二叉树的遍历。
workStealingPool具有窃取能力。
使用时最好不要加锁，而且不保证执行顺序。

6.停止线程池的正确方法

shutdown：调用了shutdown()方法不一定会立即停止，这个方法仅仅是初始整个关闭过程。因为线程池中的线程有可能正在运行，并且队列中也有待处理的任务，不可能说停就停。所以每当调用该方法时，线程池会把正在执行的任务和队列中等待的任务都执行完毕再关闭，并且在此期间如果接收到新的任务会被拒绝。

/**
 * 演示关闭线程池
 */
public class ShutDown {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1000; i++) {
            executorService.execute(new ShutDownTask());
        }
        Thread.sleep(1500);
        executorService.shutdown();

        //再次提交任务
        executorService.execute(new ShutDownTask());
    }
}

class ShutDownTask implements Runnable {

    @Override
    public void run() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName());
    }
}

isShutdown：可以用于判断线程池是否被shutdown了

/**
 * 演示关闭线程池
 */
public class ShutDown {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1000; i++) {
            executorService.execute(new ShutDownTask());
        }
        Thread.sleep(1500);
        System.out.println(executorService.isShutdown());
        executorService.shutdown();
        System.out.println(executorService.isShutdown());
        //再次提交任务
//        executorService.execute(new ShutDownTask());
    }
}

class ShutDownTask implements Runnable {

    @Override
    public void run() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName());
    }
}

isTerminated：可以判断线程是否被完全终止了

/**
 * 演示关闭线程池
 */
public class ShutDown {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1000; i++) {
            executorService.execute(new ShutDownTask());
        }
        Thread.sleep(1500);
        System.out.println(executorService.isShutdown());
        executorService.shutdown();
        System.out.println(executorService.isShutdown());
        System.out.println(executorService.isTerminated());
        //再次提交任务
//        executorService.execute(new ShutDownTask());
    }
}

将循环的次数改为100次，并且在第一次调用isTerminated方法的地方休眠10s

awaitTermination：传入等待时间，等待时间达到时判断是否停止了，主要用于检测。

//在3s后判断线程池是否被终止，返回boolean值

shutdownNow：调用了这个方法时，线程池会立即终止，并返回没有被处理完的任务。如果需要继续执行这里的任务可以再次让线程池执行这些返回的任务。

7.任务太多，怎么拒绝？

7.1 拒绝的时机

当Executor关闭时，新提交的任务会被拒绝。
以及Executor对最大线程数和工作队列容量使用有限边界并且已经饱和时。

7.2 拒绝策略

AbortPolicy（中断策略）：直接抛出异常进行拒绝
DiscardPolicy（丢弃策略）：不会得到通知，默默的抛弃掉任务
DiscardOldestPolicy（丢弃最老的）：由于队列中存储了很多任务，这个策略会丢弃在队列中存在时间最久的任务。
CallerRunsPolicy：比如主线程给线程池提交任务，但是线程池已经满了，在这种策略下会让提交任务的线程去执行。

总结：第四种拒绝策略相对于前三种更加“机智”一些，可以避免前面三种策略产生的损失。在第四种策略下可以降低提交的速度，达到负反馈的效果。

8.使用钩子为线程池加点料（可用于日志记录）

/**
 * 演示每个任务执行的前后放钩子函数
 */
public class PauseableThreadPool extends ThreadPoolExecutor {

    private boolean isPaused;
    private final ReentrantLock lock = new ReentrantLock();
    private Condition unPaused = lock.newCondition();

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                               TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }


    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                               TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }


    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                               TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                               TimeUnit unit, BlockingQueue<Runnable> workQueue,
                               ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        lock.lock();
        try {
            while (isPaused) {
                unPaused.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    private void pause() {
        lock.lock();
        try {
            isPaused = true;
        } finally {
            lock.unlock();
        }
    }

    public void resume() {
        lock.lock();
        try {
            isPaused = false;
            //唤醒全部
            unPaused.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        PauseableThreadPool pauseableThreadPool = new PauseableThreadPool(10, 20, 10L,
                TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                System.out.println("我被执行");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        for (int i = 0; i < 10000; i++) {
            pauseableThreadPool.execute(runnable);
        }
        Thread.sleep(1500);
        pauseableThreadPool.pause();
        System.out.println("线程池被暂停了");
        Thread.sleep(1500);
        pauseableThreadPool.resume();
        System.out.println("线程池被恢复了");
    }
}

9.线程池实现原理

9.1 线程池组成部分

线程池管理器
工作线程
任务队列
任务

9.2 Executor家族

Executor：它是一个顶层接口，其他接口以及类都i继承或实现于它，包含以下方法： void execute(Runnable command);
ExecutorService：它继承于Executor，是Executor的子接口，在接口内部增加了一些新的方法，例如第6小节讲到的几个方法
Executors：这个类是一个工具类，里面包含一些创建线程池的方法

9.3 线程池实现任务复用的原理

利用相同线程执行不同任务

源码分析

public void execute(Runnable command) {
    // 判断任务是否为空，为空就抛出异常
    if (command == null)
        throw new NullPointerException();
    
    int c = ctl.get();
    // 如果当前线程数小于核心线程数，就增加Worker
    if (workerCountOf(c) < corePoolSize) {
        // command就是任务，点击addWorker方法
        // 第二个参数用于判断当前线程数是否小于核心线程数
        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);
}

因为要查看的是Worker所以进入到addWorker()方法后点击Worker类查看runWorker()方法

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{

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是一个Runnable类型，调用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);
    }
}

总结：核心原理就是获取到task,如果task不为空就调用run()方法，这样就实现了线程的复用，达到让相同的线程执行不同任务的目的。

10.线程池状态

RUNNING：接受新任务并处理排队任务
SHUTDOWN：不接受新的任务但是处理排队任务
STOP：不接受新的任务，也不处理排队的任务，并中断正在执行的任务，就是调用shutdownNow()带来的效果
TIDYING：中文意思是整洁，意思就是说任务都已经终止，workerCount为零时，线程会转换到TIDYING状态，并将运行terminate()钩子方法
TERMINATED：terminate()运行完成

11.使用线程池的注意点

避免任务的堆积（堆积容易产生内存溢出）
避免线程数过多增加（缓存线程池会导致线程数过度增加）
排查线程泄漏（线程已经执行完毕却无法被回收）

作者：Oo鲁毅oO
链接：https://juejin.im/post/5e1b1fcce51d454d3046a3de

上一篇： Spring ThreadPoolTaskExecutor线程池corePoolSize vs maxPoolSize
下一篇：面试突击29:说一下线程池7个参数的含义?

网站首页 > 技术文章正文

学习线程池掌握治理线程的法宝讲讲线程池的实现原理

1.为什么需要线程池

2.什么场景下适合使用线程池

3.线程池参数介绍以及特点

3.1 corePoolSize和maxPoolSize

3.2 添加线程的规则

3.3 增减线程的特点

3.4 keepAliveTime

3.5 threadFactory

3.6 workQueue

4.线程池应该手动创建还是自动创建

4.1 直接调用JDK封装好的线程池会带来的问题

4.2 演示newFixedThreadPool内存溢出问题

4.3 newSingleThreadExecutor

4.4 newCachedThreadPool

4.5 newScheduledThreadPool

4.6 正确的创建线程池的方法

4.7 线程池里的线程数量设置多少比较合适？

5.对比线程池的特点

5.1 为什么FixedThreadPool和SingleThreadExecutor的Queue是LinkedBlockingQueue？

5.2 为什么CachedThreadPool使用的Queue是SynchronousQueue？

5.3 为什么ScheduledThreadPool使用延迟队列DelayedWorkQueue？

5.4 JDK1.8中加入的workStealingPool

6.停止线程池的正确方法

7.任务太多，怎么拒绝？

7.1 拒绝的时机

7.2 拒绝策略

8.使用钩子为线程池加点料（可用于日志记录）

9.线程池实现原理

9.1 线程池组成部分

9.2 Executor家族

9.3 线程池实现任务复用的原理

10.线程池状态

11.使用线程池的注意点

猜你喜欢

本文暂时没有评论，来添加一个吧(●'◡'●)

取消回复欢迎你发表评论:

网站首页 > 技术文章 正文

学习线程池 掌握治理线程的法宝 讲讲线程池的实现原理

1.为什么需要线程池

2.什么场景下适合使用线程池

3.线程池参数介绍以及特点

3.1 corePoolSize和maxPoolSize

3.2 添加线程的规则

3.3 增减线程的特点

3.4 keepAliveTime

3.5 threadFactory

3.6 workQueue

4.线程池应该手动创建还是自动创建

4.1 直接调用JDK封装好的线程池会带来的问题

4.2 演示newFixedThreadPool内存溢出问题

4.3 newSingleThreadExecutor

4.4 newCachedThreadPool

4.5 newScheduledThreadPool

4.6 正确的创建线程池的方法

4.7 线程池里的线程数量设置多少比较合适？

5.对比线程池的特点

5.1 为什么FixedThreadPool和SingleThreadExecutor的Queue是LinkedBlockingQueue？

5.2 为什么CachedThreadPool使用的Queue是SynchronousQueue？

5.3 为什么ScheduledThreadPool使用延迟队列DelayedWorkQueue？

5.4 JDK1.8中加入的workStealingPool

6.停止线程池的正确方法

7.任务太多，怎么拒绝？

7.1 拒绝的时机

7.2 拒绝策略

8.使用钩子为线程池加点料（可用于日志记录）

9.线程池实现原理

9.1 线程池组成部分

9.2 Executor家族

9.3 线程池实现任务复用的原理

10.线程池状态

11.使用线程池的注意点

猜你喜欢

本文暂时没有评论，来添加一个吧(●'◡'●)

取消回复欢迎 你 发表评论:

网站首页 > 技术文章正文

学习线程池掌握治理线程的法宝讲讲线程池的实现原理

取消回复欢迎你发表评论: