JUC之线程池基础与简单源码分析

线程池

定义和方法

线程池的工作时控制运行的线程数量,处理过程中将任务放入队列,然后在线程创建后启动这些任务,如果线程数量超过了最大数量,超出数量的线程排队等候,等待其他线程执行完成,再从队列中取出任务来执行。

特点:

线程复用,控制最大并发数,管理线程。

好处:

  1. 降低资源消耗。通过重复利用已创建的线程来降低线程创建和销毁造成的消耗。
  2. 提升响应速度。当任务到达时,任务不需要等待线程创建就能立即执行
  3. 提高线程的可管理性。当线程时稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配、调优和监控。

简单的架构图:

JUC之线程池基础与简单源码分析

我们使用的其实就是ThreadPoolExecutor.

阿里巴巴开发规范手册:

【强制】线程池不允许使用 Executors 去创建,而是通过 ThreadPoolExecutor 的方式,这样的处理方式让写的同学更加明确线程池的运行规则,规避资源耗尽的风险。

说明:Executors 返回的线程池对象的弊端如下:

1)FixedThreadPool 和 SingleThreadPool:

允许的请求队列长度为 Integer.MAX_VALUE,可能会堆积大量的请求,从而导致 OOM。

2)CachedThreadPool:

允许的创建线程数量为 Integer.MAX_VALUE,可能会创建大量的线程,从而导致 OOM。

首先使用工具类(Executors)来创建线程池:

FixedThreadPool被称为可重用固定线程数的线程池,执行长期任务性能好,创建一个线程池,一池有N个固定的线程,有固定的线程数的线程池。

public class demo1 {
    public static void main(String[] args) {
        //使用Executors工具类来创建newFixedThreadPool线程池
        ExecutorService executor = Executors.newFixedThreadPool(3);

        try{
            for (int i = 0; i < 10; i++) {
                executor.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"任务执行");
                });
            }
        } finally {
            executor.shutdown();
        }
    }
}

SingleThreadExecutor是使用单个worker线程的Executor,省去了创建线程和销毁线程时资源消耗。

public class demo1 {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        try{
            for (int i = 0; i < 10; i++) {
                executor.execute(()->{

                    System.out.println(Thread.currentThread().getName()+"\t任务执行");
                });
            }
        } finally {
            executor.shutdown();
        }
    }
}

CachedThreadPool是一个会根据需要创建新线程的线程池。可扩容的线程池。

public class demo1 {
    public static void main(String[] args) {
        //使用Executors工具类来创建newFixedThreadPool线程池
        //ExecutorService executor = Executors.newFixedThreadPool(3);
        //一个线程池就一个线程
        //ExecutorService executor = Executors.newSingleThreadExecutor();
        ExecutorService executor = Executors.newCachedThreadPool();
        try{
            for (int i = 0; i < 10; i++) {
                executor.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"\t任务执行");
                });
            }
        } finally {
            executor.shutdown();
        }
    }
}
pool-1-thread-2	任务执行
pool-1-thread-1	任务执行
pool-1-thread-4	任务执行
pool-1-thread-3	任务执行
pool-1-thread-5	任务执行
pool-1-thread-6	任务执行
pool-1-thread-7	任务执行
pool-1-thread-8	任务执行
pool-1-thread-9	任务执行
pool-1-thread-10	任务执行

当设置延时操作来模仿耗时的代码时使用线程池的过程。

package com.JucPool;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 基础认识以及线程池的三大方法
 */
public class demo1 {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        try{
            for (int i = 0; i < 10; i++) {
                //暂停几秒
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                executor.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"\t任务执行");
                });
            }
        } finally {
            executor.shutdown();
        }
    }
}
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行
pool-1-thread-1	任务执行

Process finished with exit code 0

这时就降为SingleThreadExecutor线程池。

简单源码分析

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

通过对比发现,底层都是通过ThreadPoolExecutor来实现,只是参数的不同,传递的参数与阻塞队列也有关系。

参数:(非常重要)

  1. ·corePool:线程池常驻核心线程池的大小。
  2. ·maximumPool:最大线程池的大小--线程池中能够容纳同时执行的最大线程数
  3. keepAliveTime:多余的空闲线程的存活时间,当前池中线程数超过了corePool时并且空闲时间到达keepAliveTime,多余的线程会被销毁直到只剩下corePool个线程为止
  4. unit – keepAliveTime参数的时间单位
  5. workQueue – 用于在执行任务之前保存任务的队列。 此队列将仅保存由execute方法提交的Runnable任务。

在ThreadPoolExecutor中还存在两个参数:

  1. ThreadFactory: 表示生成线程池中工作线程的线程工厂,用于创建线程,一般默认
  2. Handler:拒绝策略,表示当队列满了,并且工作的线程要超过最大线程池的大小,如何让拒绝请求执行的runnable的策略

上述是线程池中的七大参数总结,很重要,关于什么情况下走到拒绝策略,后面会分析线程池的执行流程。

上一篇:ZooKeeper : Curator框架之分布式屏障DistributedBarrier


下一篇:JUC 中 4 个常用的并发工具类