java多线程解析之线程池

首先分析下为什么需要使用线程池?

假如不使用线程池,可能会造成线程数量过大,程序崩溃。因为线程也需要占用内存与CPU资源。

如果是你,会如何设计线程池?

最简单的,我们会使用一个容器,存放一定数量的线程,超过数量则不创建。那么,我们来看看jdk是怎么处理的。

线程池构造函数分析

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

1.corePoolSize:核心线程数
2.maximumPoolSize:最大线程数
3.keepAliveTime和unit:超过核心线程数的线程的空闲存活时间,后面是时间的单位
4.workQueue:当前线程数达到核心线程数之后,会通过workQueue接受任务进入等待
5.threadFactory:创建线程的工厂
6.handler:当workQueue满了并且已经达到最大线程数的拒绝策略

异步任务进入线程池执行流程

线程池执行任务核心源码分析

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        // 获取线程池的状态
        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);
    }

线程池如何保证核心线程数不销毁

当线程池执行任务时,会调用getTask获取需要执行的任务,而workQueue.take()就会阻塞住,直到队列中有任务到达。

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进入任务
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

如何使用线程池

1.Executors.newFixedThreadPool (创建固定线程池)

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        for (int i = 0; i < 3; i++) {
            executorService.submit(() -> {
                try {
                    Thread.sleep(3000);
                    System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639356608206
当前线程:pool-1-thread-1,当前时间戳:1639356611215
当前线程:pool-1-thread-1,当前时间戳:1639356614219

构造函数分析:

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

线程池的核心线程数和最大线程数都为指定线程数,无线程增长,内部用LinkedBlockingQueue无界阻塞队列接受任务,可能会造成任务一直在阻塞中,无法执行。

2.Executors.newSingleThreadExecutor (创建单个线程)

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 3; i++) {
            executorService.submit(() -> {
                try {
                    Thread.sleep(3000);
                    System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639367551803
当前线程:pool-1-thread-1,当前时间戳:1639367554805
当前线程:pool-1-thread-1,当前时间戳:1639367557807

构造函数分析:

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

核心线程数和最大线程数都为1,无线程增长。内部使用LinkedBlockingQueue无界阻塞队列接受任务,可能会造成任务一直在阻塞中,无法执行。

3.Executors.newCachedThreadPool (创建可缓存使用的线程池)

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 3; i++) {
            executorService.submit(() -> {
                try {
                    Thread.sleep(3000);
                    System.out.println("当前线程:" + Thread.currentThread().getId() + ",当前时间戳:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }

构造函数分析:

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

核心线程数为0,最大线程数为最大整数,存活时间60秒,该线程池可能会一直创建线程,导致内存溢出。

4.newSingleThreadScheduledExecutor (创建单个定时任务线程池)

   public static void main(String[] args) {
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        // 首次0延时执行,之后每三秒执行一次
        executorService.scheduleAtFixedRate(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
        }, 0, 3, TimeUnit.SECONDS);
    }
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639369721662
当前线程:pool-1-thread-1,当前时间戳:1639369724667
当前线程:pool-1-thread-1,当前时间戳:1639369727670
当前线程:pool-1-thread-1,当前时间戳:1639369730669

构造函数分析

    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }

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

底层通过DelegatedScheduledExecutorService实现,核心线程数为1,最大线程数为最大整数,通过DelayedWorkQueue延时队列实现定时调度功能。

线程池有哪些拒绝策略

1.ThreadPoolExecutor.AbortPolicy (抛出拒绝异常)

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(1, 3,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.AbortPolicy());
        for (int i = 0; i < 8; i++) {
            executorService.submit(() -> {
                System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
            });
        }
    }
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-3,当前时间戳:1639399896121
当前线程:pool-1-thread-2,当前时间戳:1639399896121
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@58372a00 rejected from java.util.concurrent.ThreadPoolExecutor@4dd8dc3[Running, pool size = 3, active threads = 3, queued tasks = 2, completed tasks = 0]

当达到最大线程数并且队列满了之后,就会调用拒绝策略,抛出异常,符合预期。

2.ThreadPoolExecutor.CallerRunsPolicy(使用调用者线程执行)

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(1, 3,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 8; i++) {
            executorService.submit(() -> {
                System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
            });
        }
    }
结果打印:
当前线程:main,当前时间戳:1639400060907
当前线程:pool-1-thread-3,当前时间戳:1639400060907
当前线程:pool-1-thread-1,当前时间戳:1639400060907
当前线程:pool-1-thread-2,当前时间戳:1639400060907
当前线程:pool-1-thread-3,当前时间戳:1639400060908
当前线程:pool-1-thread-1,当前时间戳:1639400060908
当前线程:main,当前时间戳:1639400060907
当前线程:pool-1-thread-2,当前时间戳:1639400060908

当达到最大线程数并且队列满了之后,就会使用调用者线程main执行。

3.ThreadPoolExecutor.DiscardPolicy (直接丢弃任务)

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(1, 3,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.DiscardPolicy());
        for (int i = 0; i < 8; i++) {
            int no = i;
            executorService.submit(() -> {
                System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
            });
        }
    }
结果打印:
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400376316
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400376316
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400376316
当前循环:2当前线程:pool-1-thread-2,当前时间戳:1639400376316
当前循环:1当前线程:pool-1-thread-1,当前时间戳:1639400376316

当达到最大线程数并且队列满了之后,直接丢弃之后的任务,靠前的任务能保证执行。

4.ThreadPoolExecutor.DiscardOldestPolicy (丢弃最早的任务)

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(1, 3,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.DiscardOldestPolicy());
        for (int i = 0; i < 8; i++) {
            int no = i;
            executorService.submit(() -> {
                System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
            });
        }
    }
结果打印:
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400217828
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400217828
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400217828
当前循环:7当前线程:pool-1-thread-2,当前时间戳:1639400217828
当前循环:6当前线程:pool-1-thread-1,当前时间戳:1639400217828

当达到最大线程数并且队列满了之后,直接丢弃最早的任务,靠后的任务能保证执行。

5.自定义拒绝策略

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(1, 3,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                (r, executor) -> {
                    System.out.println("执行自定义拒绝策略");
                });
        for (int i = 0; i < 8; i++) {
            int no = i;
            executorService.submit(() -> {
                System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
            });
        }
    }
结果打印:
执行自定义拒绝策略
执行自定义拒绝策略
执行自定义拒绝策略
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400497694
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400497694
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400497694
当前循环:1当前线程:pool-1-thread-1,当前时间戳:1639400497695
当前循环:2当前线程:pool-1-thread-3,当前时间戳:1639400497695
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容