Java并发 - 线程池

水深无声 2022-09-07 01:18 238阅读 0赞

# Java并发 - 线程池 #

务必牢记：三大方法、七个参数、四个策略。

池化技术：简单点来说，\*\*就是提前保存大量的资源，以备不时之需。\*\*池化技术能够减少资源对象的创建次数。每一次创建一个线程都需要申请一块堆内存空间，用完之后销毁，都会涉及到系统的调用，非常消耗CPU资源。

线程池优点：**线程复用**、**可以控制最大并发数**、**管理线程**、提高响应速度、降低资源消耗、方便管理。

### 创建线程池 ###

1、Executors

/** * 阿里开发手册中写到：不允许使用 Executors 去创建线程，而是通过 ThreadPoolExecutor 的 * 处理方式来写的同学更加明确线程池的运行规则，规避资源耗尽的风险。 * Executors返回的线程池对象的弊端： * （1）newFixedThreadPool 和 newSingleThreadExecutor： * 允许的请求队列为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM（内存溢出） * （2）CachedThreadPool 和 ScheduledThreadPool: * 允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。 */
    public void testExecutors() { 
        // 单个线程
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        // 指定线程数
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        // 可伸缩
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) { 
            cachedThreadPool.execute(() -> { 
                System.out.println(Thread.currentThread().getName() + " => ok");
            });
        }
        cachedThreadPool.shutdown();
    }

对应池的构造方法：

public static ExecutorService newSingleThreadExecutor() { 
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    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>());
    }
    // 其实按理里面还有一个方法，用于周期性任务的线程池

Executors 工具类只是配置了几个参数的线程池，本质都是 new ThreadPoolExecutor

public ThreadPoolExecutor(int corePoolSize,   // 核心线程数
                              int maximumPoolSize,// 最大核心线程数
                              long keepAliveTime, // 未被调用超时后释放
                              TimeUnit unit, 	  // 超时单位 
                              BlockingQueue<Runnable> workQueue, // 阻塞队列
                              ThreadFactory threadFactory,       // 线程工厂 
                              RejectedExecutionHandler handler)  // 拒绝策略

**corePoolSize**

corePoolSize the number of threads to keep in the pool, even if they are idle, unless \{@code allowCoreThreadTimeOut\} is set。

线程池中一直存在的线程数量，如果设置 allowCoreThreadTimeOut 为 true时，将会释放全部空闲线程。

**workQueue**

the queue to use for holding tasks before they are executed. This queue will hold only the \{@code Runnable\} tasks submitted by the \{@code execute\} method.

当 corePoolSize 线程满了之后，其它线程将会放到阻塞队列中。

**maximumPoolSize**

当 corePoolSize 和 workQueue 都满了之后，如果新增的线程数+核心线程数之后达到了 maximumPoolSize 的数量，将会触发拒绝策略。

**keepAliveTime**

如果此时线程池中的线程数量超过了 corePoolSize 的数量，也就是 maximumPoolSize减去corePoolSize 的线程在 keepAliveTime 时间内未被调用就会被释放。

**RejectedExecutionHandler**

达到 maximumPoolSize 后触发拒绝策略，RejectedExecutionHandler 接口有四个实现类：

/** * A handler for rejected tasks that runs the rejected task * directly in the calling thread of the {@code execute} method, * unless the executor has been shut down, in which case the task * is discarded. * 会在提交任务的线程中执行，不在提交目标中执行。 * 例如主线程中提交一个任务到线程池中，该任务不会在线程池中执行，会在主线程中执行 */
    public static class CallerRunsPolicy implements RejectedExecutionHandler { }
    
    /** * A handler for rejected tasks that discards the oldest unhandled * request and then retries {@code execute}, unless the executor * is shut down, in which case the task is discarded. * 尝试和最早执行的任务竞争，竞争成功执行任务，失败则放弃执行 */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler { }
    
    /** * A handler for rejected tasks that silently disca rds the * rejected task. * 放弃执行任务，不会抛出异常 */	
    public static class DiscardPolicy implements RejectedExecutionHandler { }
    
    /** * A handler for rejected tasks that throws a * 放弃执行任务，抛出异常 */
    public static class AbortPolicy implements RejectedExecutionHandler { }

**ThreadFactory**

线程池的threadFactory入参描述：the factory to use when the executor creates a new thread。

public interface ThreadFactory { 
    
        /** * Constructs a new {@code Thread}. Implementations may also initialize * priority, name, daemon status, {@code ThreadGroup}, etc. * * @param r a runnable to be executed by new thread instance * @return constructed thread, or {@code null} if the request to * create a thread is rejected */
        Thread newThread(Runnable r);
    }

从注释中可知，通过ThreadFactory的newThread方法，可以对创建的线程做一些操控：

*  指定优先级
 *  指定线程名称，方便监控
 *  指定是否守护线程，守护线程不阻塞进程
 *  指定线程组

> ### 在创建线程池的时候，maxiMumPoolSize该如何设置多少合适? ###

// 获取CPu的核数
    Runtime.getRuntime().availableProcessors();

**IO密集型**（网络操作、数据库操作、文件操作）：判断程序中十分消耗 IO 的线程数，例如当前要处理15个大型任务，至少得15个线程去执行，那么 maximumPoolSize的值一般设置为当前所需线程数的两倍。tomcat 中默认最大线程数为 200。

**计算型任务**（CPU密集型）：一般设置为CPU数量的1-2倍。几核CPU，就是几的话可以保持CPU的效率最高。

还有一种方法就是配置后运行实验，查看CPU的利用率，达到80%则表示被充分利用。

> **如何计算线程池中的最大承载数？**  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjQ3OTk5_size_16_color_FFFFFF_t_70]

这里有一个误区，有人会认为队列是保存的一批核心线程数，实际上队列的容量就是排队的线程数量。而计算最大的承载数，就是当前的最大线程数+队列的容量。

> **无法预估corePoolSize和maximumPoolSize，变化数比较多时，应该如何设置参数？**

/* 每个任务需要tasktime秒处理，则每个线程每秒可处理1/tasktime个任务。系统每秒有tasks个任务需要处理，则需要的线程数为：tasks/(1/tasktime)。 即tasks*tasktime个线程数。假设系统每秒任务数为100到1000之间，每个任务耗时0.1秒，则需要100x0.1至1000x0.1，即10到100个线程。那么corePoolSize应该设置为大于10。 具体数字最好根据8020原则，即80%情况下系统每秒任务数，若系统80%的情况下任务数小于200，最多时为1000，则corePoolSize可设置为20。 */
    corePoolSize    	0
    /* 当系统负载达到最大值时，核心线程数已无法按时处理完所有任务，这时就需要增加线程。每秒200个任务需要20个线程，那么当每秒达到1000个任务时，则需要(1000-queueCapacity)*(20/200)，即60个线程，可将maxPoolSize设置为60。 */
    maximumPoolSize  	Integer.MAX_VALUE 
     /* 线程数量只增加不减少也不行。当负载降低时，可减少线程数量，如果一个线程空闲时间达到keepAliveTiime，该线程就退出。默认情况下线程池最少会保持corePoolSize个线程。keepAliveTiime设定值可根据任务峰值持续时间来设定。 以上关于线程数量的计算并没有考虑CPU的情况。若结合CPU的情况，比如，当线程数量达到50时，CPU达到100%，则将maxPoolSize设置为60也不合适，此时若系统负载长时间维持在每秒1000个任务，则超出线程池处理能力，应设法降低每个任务的处理时间(tasktime)。 */
    keepAliveTime		60L
    unit 			    TimeUnit.SECONDS
    workQueue			new SynchronousQueue<>() // 使用同步队列
    allowCoreThreadTimeout true                  // 既然无法预估线程数，核心线程数超时不活跃会自动关闭

这种方式也适用于快速执行完任务。

## 补充内容 ##

> 线程池中，是不是线程越多越好？为什么？

1、线程在java中是一个对象，更是操作系统的资源，线程创建、销毁需要时间。如果创建时间+销毁时间>执行任务时间就很不合算。

2、java对 象占用堆内存，操作系统线程占用系统内存，根据jvm规范，一个线程默认最大栈大小1M，这个栈空间是需要从系统内存中分配的。线程过多，会消耗很多的内存。

3、操作系统需要频繁切换线程上下文(大家都想被运行)，影响性能。

线程池的推出，就是为了控制线程数。

> 线程池API

<table> 
 <thead> 
  <tr> 
   <th>类型</th> 
   <th>名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>接口</td> 
   <td>Executor</td> 
   <td>最上层的接口，定义了执行任务的方法execute</td> 
  </tr> 
  <tr> 
   <td>接口</td> 
   <td>ExecutorService</td> 
   <td>继承了Executor接口，拓展了Callable、Future、 关闭方法</td> 
  </tr> 
  <tr> 
   <td>接口</td> 
   <td>ScheduleExecutorService</td> 
   <td>继承了ExecutorService,增加了定时任务相关的方法</td> 
  </tr> 
  <tr> 
   <td>实现类</td> 
   <td>ThreadPoolExecutor</td> 
   <td><strong>基础、标准的线程池实现</strong></td> 
  </tr> 
  <tr> 
   <td>实现类</td> 
   <td>ScheduleThreadPoolExecutor</td> 
   <td>继承了ThreadPoolExecutor,实现了ScheduledExecutorService中相关定时任务的方法</td> 
  </tr> 
 </tbody> 
</table>

ExecutorService 方法定义

//监测ExecutorService是否已经关闭，直到所有任务完成执行，或超时发生，或当前线程被中断
    awaitTermination(long timeout, TimeUnit unit)
    //执行给定的任务集合，执行完毕后，返回结果
    invokell(Collection<? extends Callable<T>> tasks)
    //执行给定的任务集合，执行完毕或者超时后，返回结果，其他任务终止
    invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
    //执行给定的任务，任意一个执行成功则返回结果，其他任务终止
    invokeAny(Collection<? extends Callable<T>> tasks)
    //执行给定的任务，任意-个执行成功或者超时后，则返回结果，其他任务终止
    invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
    //如果此线程池已关闭，则返回true。
    isShutdown()
    //如果关闭后所有任务都已完成，则返回true。
    is Terminated()
    //优雅关闭线程池，之前提交的任务将被执行，但是不会接受新的任务。
    shutdown()
    //尝试停止所有正在执行的任务，停止等待任务的处理，并返回等待执行任务的列表。
    shutdownNow()
    //提交一个用于执行的Callable返回任务，并返回一个Future,用于获取Callable执行结果
    submit(Callable<T> task)
    //提交可运行任务以执行，并发回一个Future对象，执行结果为null
    submit(Runnable task)
    //提交可运行任务以执行，并返回Future,执行结果为传入的result
    submit(Runnable task, T result)

ScheduledExecutorService 方法定义

// 创建并执行一个一次性任务
    schedule(Runnable command, long delay, TimeUnit unit);
    schedule(Callable<V> callable,long delay, TimeUnit unit);
    // 如果一次任务执行时长超过了周期时间，下一次任务会等到该次任务执行结束后，立刻执行
    scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);
    // 如果一次任务执行时长超过了周期时间，下一次任务会在该次任务执行结束的基础时间上计算执行的延时
    scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjQ3OTk5_size_16_color_FFFFFF_t_70]: /images/20220829/0b2ebff6b655421981b0d750b382b93e.png