让您轻松理解java线程池

Posted 2022-12-07 请叫我东子

1.线程

线程是调度cpu的最小单元，也叫轻量级的进程。

2.两种线程模型

• 用户级线程（ULT）：指不需要内核支持而在用户程序中实现的线程，它的内核的切换是由用户态程序自己控制内核的切换，不需要内核的干涉。但是它不能像内核级线程一样更好的运用多核CPU。
• 内核级线程（KLT）：切换由内核控制，当线程进行切换的时候，由用户态转化为内核态。切换完毕要从内核态返回用户态。可以很好的运用多核CPU，就像Windows电脑的四核八线程，双核四线程一样。

3.线程池

• 线程池管理器（ThreadPool）：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；
• 工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；
• 任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
• 任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。

4.常用线程池

类型	说明
SingleThreadExecutor	单一线程的线程池
FixedThreadPool	固定大小的线程池
CachedThreadPool	可缓冲的线程池
ScheduledThreadPool	无限制大小线程池，定时场景

4.1 SingleThreadExecutor

单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序执行。

public class singleThreadExecutor 
    public static void main(String[] args) 
        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(new Runnable() 
            @Override
            public void run() 
                for (int i = 0; i < 10; i++) 
                    System.out.print(1 + " ");
                
            
        );

        executor.execute(new Runnable() 
            @Override
            public void run() 
                for (int i = 0; i < 10; i++) 
                    System.out.print(2 + " ");
                
            
        );
        executor.execute(new Runnable() 
            @Override
            public void run() 
                for (int i = 0; i < 10; i++) 
                    System.out.print(3 + " ");
                
            
        );

        executor.shutdown();
    

4.2 CachedThreadPool

可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

public class CachedThreadPool 
    public static void main(String[] args) 
        ExecutorService executor = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) 
            final int index = i;
            try 
                Thread.sleep(1000);
             catch (InterruptedException e) 
                e.printStackTrace();
            
            executor.execute(new Runnable() 
                public void run() 
                    System.out.println(index);
                
            );
        
        executor.shutdown();//关闭线程
    
 

4.3 FixedThreadPool

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待.
定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()

public class FixedThreadPool 
    public static void main(String[] args) 
      //因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。
        ExecutorService executor = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 10; i++) 
            final int index = i;
            executor.execute(new Runnable() 
                public void run() 
                    try 
                        System.out.println(index);
                        Thread.sleep(1000);
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
            );
        
        executor.shutdown();
    

4.4 ScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

public class ScheduledThreadPool 
    public static void main(String[] args) 
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
        scheduledThreadPool.schedule(new Runnable() 
            public void run() 
                System.out.println("delay 3 seconds");
            
        , 3, TimeUnit.SECONDS);//等待三秒执行

        scheduledThreadPool.scheduleAtFixedRate(new Runnable() 
            public void run() 

                System.out.println("delay 1 seconds, and excute every 3 seconds");
            

        , 1, 3, TimeUnit.SECONDS);//表示延迟1秒后每3秒执行一次。

    

5. ThreadPoolExecutor+BlockingQueue使用详解

5.1 使用场景

• 需要的子线程数量很多，但是数量不确定。
• 子线程有自己的优先级，根据优先级来确定执行的先后顺序。
• 监听线程池的开始，结束，关闭等状态。

5.2 ThreadPoolExecutor构造方法

//使用给定的初始参数和默认线程工厂以及拒绝的执行处理程序创建新的ThreadPoolExecutor。
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue)

//使用给定的初始参数和默认拒绝执行处理程序创建新的ThreadPoolExecutor
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory)
//使用给定的初始参数和默认线程工厂创建一个新的code ThreadPoolExecutor
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler)
//使用给定的初始参数创建一个新的ThreadPoolExecutor
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

主要参数

• corePoolSize：核心线程数
  核心线程会一直存活，及时没有任务需要执行。当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理。
• maxPoolSize：最大线程数
  当线程数>=corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务。
  当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常。
• keepAliveTime：非核心线程闲置时的超时时长超过这个时长，非核心线程就会被回收。当ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true 时，keepAliveTime 同样会作用于核心线程。
• unit：用于指定 keepAliveTime 参数的时间单位
  常用的有 TimeUnit .MILLISECONDS 和TimeUnit .SECONDS。
• workQueue：线程池中的任务队列通过线程池的 execute方法提交的Runnable 对象会存储在这个参数中。
• threadFactory：线程工厂为线程池提供创建新的线程的功能。threadFactory 是一个接口，它只有一个方法: public abstract Thread newThread (Runnable r);
• RejectedExecutionHandler：通常叫做拒绝策略在线程池已经关闭的情况下或者当线程数已经达到maxPoolSize且队列已满。只要满足其中一种时，在使用execute()来提交新的任务时将会拒绝，而默认的拒绝策略是抛一个RejectedExecutionException 异常。

5.3 ThreadPoolExecutor执行顺序

1. 当线程数小于核心线程数时，创建线程。
2. 当线程数大于等于核心线程数，且任务队列未满时，将任务放入任务队列。
3. 当线程数大于等于核心线程数，且任务队列已满。
   若线程数小于最大线程数，创建线程。
   若线程数等于最大线程数，抛出异常，拒绝任务 。

5.4 BlockingQueue

BlockingQueue是一个特殊的队列，当我们从BlockingQueue中取数据时，如果BlockingQueue是空的，
则取数据的操作会进入到阻塞状态，当 BlockingQueue 中有了新数据时，这个取数据的操作又会被重新唤醒。
同理，如果 BlockingQueue 中的数据已经满了，往BlockingQueue 中存数据的操作又会进入阻塞状态，直到 BlockingQueue 中又有新的空间，存数据的操作又会被重新唤醒。它的泛型限定它是用来存放 Runnable对象的。

5.5 几种常用的BlockingQueue

• ArrayBlockingQueue：
  这个表示一个规定了大小的 BlockingQueue，ArrayBlockingQueue 的构造函数接受一个int类型的数据，该数据表示BlockingQueue 的大小，存储在 ArrayBlockingQueue 中的元素按照 FIFO（先进先出）的方式来进行存取。
• LinkedBlockingQueue：
  这个表示一个大小不确定的 BlockingQueue，在LinkedBlockingQueue 的构造方法中可以传一个 int 类型的数据，这样创建出来的 LinkedBlockingQueue是有大小的，默认LinkedBlockingQueue的大小就为 Integer.MAX_VALUE。
• PriorityBlockingQueue：
  这个队列和 LinkedBlockingQueue 类似，不同的是PriorityBlockingQueue 中的元素不是按照 FIFO来排序的，而是按照元素的Comparator来决定存取顺序的（这个功能也反映了存入 PriorityBlockingQueue中的数据必须实现了 Comparator 接口）。
• SynchronousQueue：
  这个是同步 Queue，属于线程安全的 BlockingQueue的一种，在 SynchronousQueue 中，生产者线程的插入操作必须要等待消费者线程的移除操作，Synchronous内部没有数据缓存空间，因此我们无法对 SynchronousQueue进行读取或者遍历其中的数据，元素只有在你试图取走的时候才有可能存在。我们可以理解为生产者和消费者互相等待，等到对方之后然后再一起离开。

5.6 常用线程池对应的BlockingQueue

• newFixedThreadPool()—>LinkedBlockingQueue
• newSingleThreadExecutor()—>LinkedBlockingQueue
• newCachedThreadPool()—>SynchronousQueue
• newScheduledThreadPool()—>DelayedWorkQueue

使用代码示例

public class ThreadPoolExecutorDemo 
    public static void main(String[] args) 

        BlockingQueue bq = new ArrayBlockingQueue(5);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 30, 10, TimeUnit.SECONDS, bq);
        executor.allowCoreThreadTimeOut(true);
        for (int i = 0; i < 30; i++) 
            executor.execute(new PrintDate());
        
    

    public static class PrintDate implements Runnable 
        @Override
        public void run() 
            String threadName = Thread.currentThread().getName();
            System.out.println(threadName + ":" + System.currentTimeMillis());
        
    

如何设置参数
默认值

corePoolSize=1
queueCapacity=Integer.MAX_VALUE
maxPoolSize=Integer.MAX_VALUE
keepAliveTime=60s
allowCoreThreadTimeout=false
rejectedExecutionHandler=AbortPolicy()

如何来设置

tasks ：每秒的任务数，假设为500~1000
taskcost：每个任务花费时间，假设为0.1s
responsetime：系统允许容忍的最大响应时间，假设为1s

• corePoolSize = 每秒需要多少个线程处理？
• threadcount = tasks/(1/taskcost) =tasks·taskcout = (500~1000)·0.1 = 50~100 个线程。corePoolSize设置应该大于50
  根据8020原则，如果80%的每秒任务数小于800，那么corePoolSize设置为80即可
• queueCapacity = (coreSizePool/taskcost)responsetime
  计算可得 queueCapacity = 80/0.11 = 80。意思是队列里的线程可以等待1s，超过了的需要新开线程来执行
  切记不能设置为Integer.MAX_VALUE，这样队列会很大，线程数只会保持在corePoolSize大小，当任务陡增时，不能新开线程来执行，响应时间会随之陡增。
• maxPoolSize = (max(tasks)- queueCapacity)/(1/taskcost)
  计算可得 maxPoolSize = (1000-80)/10 = 92
  最大任务数-队列容量）/每个线程每秒处理能力 = 最大线程数
• rejectedExecutionHandler：根据具体情况来决定，任务不重要可丢弃，任务重要则要利用一些缓冲机制来处理
• keepAliveTime和allowCoreThreadTimeout采用默认通常能满足。

到这里线程池的基本使用就差不多了。