java多线程解析之线程池

作者: java自力更生 | 来源:发表于2021-12-13 21:02 被阅读0次

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

假如不使用线程池，可能会造成线程数量过大，程序崩溃。因为线程也需要占用内存与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

java多线程解析之线程池

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

如果是你，会如何设计线程池？

线程池构造函数分析

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

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

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

如何使用线程池

线程池有哪些拒绝策略

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读