文章目录
- java线程池
- newCachedThreadPool:
- newFixedThreadPool:
- newScheduledThreadPool:
- newSingleThreadExecutor:
- 线程池原理
- ThreadPoolExecutor
- 工作队列BlockingQueue详解
- 实现BlockingQueue接口的常见类如下:
** CPU 密集型就采用 CPU 核数 + 1,IO 密集型就是 CPU 核数 * 2。 **
首先思考为什么大家都这么考虑?
因为这是一种不容易出错的话术。因为线程池的参数是跟据业务以及系统整体负载定义的,并没有绝对公式。 当我们定义线程池的时候,首先要从业务上判断,这是 IO 密集还是 CPU 密集。其次,要从整体系统负载上,判断线程池多不多?或者说是否有大批量消耗 CPU 的任务。
如果说项目任务比较单一,线程池较少且业务需要更快的响应时间。那么如果 IO 密集型任务,核心线程数设置 CPU 核心数 / 0.2,最大线程数设置核心线程数 * 1.5。经过我实际测试,这种能更好压榨服务器 CPU。
如果说项目任务比较多,线程池定义也比较多,那么就要考虑定义多个线程池以及很多线程之间的 CPU 上下文切换问题。因为,当你的线程数远远大于 CPU 且都在运行时,线程是拿不到 CPU 调度的。这个时候,我们就该从全局角度上考虑将线程数调整小一些。
最后总结:没有固定公式,需要考虑项目任务运行情况、线程定义的多少、线程池运行是否需要高实时等。
在执行一个异步任务或并发任务时,往往是通过直接new Thread()方法来创建新的线程,这样做弊端较多,更好的解决方案是合理地利用线程池,线程池的优势很明显,如下:
- 降低系统资源消耗,通过重用已存在的线程,降低线程创建和销毁造成的消耗;
- 提高系统响应速度,当有任务到达时,无需等待新线程的创建便能立即执行;
- 方便线程并发数的管控,线程若是无限制的创建,不仅会额外消耗大量系统资源,更是占用过多资源而阻塞系统或oom等状况,从而降低系统的稳定性。线程池能有效管控线程,统一分配、调优,提供资源使用率;
- 更强大的功能,线程池提供了定时、定期以及可控线程数等功能的线程池,使用方便简单。
java_15">java线程池
java通过Executors提供四种线程池,分别为:
newCachedThreadPool:
创建一个可缓存的无界线程池,如果线程池长度超过处理需要,可灵活回收空线程,若无可回收,则新建线程。当线程池中的线程空闲时间超过60s,则会自动回收该线程,当任务超过线程池的线程数则创建新的线程,线程池的大小上限为Integer.MAX_VALUE,可看作无限大。
newFixedThreadPool:
创建一个指定大小的线程池,可控制线程的最大并发数,超出的线程会在LinkedBlockingQueue阻塞队列中等待。
newScheduledThreadPool:
创建一个定长的线程池,可以指定线程池核心线程数,支持定时及周期性任务的执行
newSingleThreadExecutor:
创建一个单线程化的线程池,它只有一个线程,用仅有的一个线程来执行任务,保证所有的任务按照指定顺序(FIFO,LIFO,优先级)执行,所有的任务都保存在队列LinkedBlockingQueue中,等待唯一的单线程来执行任务。
线程池原理
Executors类提供4个静态工厂方法:newCachedThreadPool()、newFixedThreadPool(int)、newSingleThreadExecutor和newScheduledThreadPool(int)。这些方法最终都是通过ThreadPoolExecutor类来完成的,这里强烈建议大家直接使用Executors类提供的便捷的工厂方法,能完成绝大多数的用户场景,当需要更细节地调整配置,需要先了解每一项参数的意义。
ThreadPoolExecutor
java">public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);
}
创建线程池,在构造一个新的线程池时,必须满足下面的条件:
- corePoolSize(线程池基本大小)必须大于或等于0;
- maximumPoolSize(线程池最大大小)必须大于或等于1;
- maximumPoolSize必须大于或等于corePoolSize;
- keepAliveTime(线程存活保持时间)必须大于或等于0;
- workQueue(任务队列)不能为空;
- threadFactory(线程工厂)不能为空,默认为DefaultThreadFactory类
- handler(线程饱和策略)不能为空,默认策略为ThreadPoolExecutor.AbortPolicy。
- workQueue(任务队列):用于传输和保存等待执行任务的阻塞队列。可以使用此队列与线程池进行交互:
- 如果运行的线程数少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。
- 如果运行的线程数等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。
- 如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。
- threadFactory(线程工厂):用于创建新线程。由同一个threadFactory创建的线程,属于同一个ThreadGroup,创建的线程优先级都为Thread.NORM_PRIORITY,以及是非守护进程状态。threadFactory创建的线程也是采用new Thread()方式,threadFactory创建的线程名都具有统一的风格:pool-m-thread-n(m为线程池的编号,n为线程池内的线程编号);
- handler(线程拒绝策略):当线程池和队列都满了,则表明该线程池已达饱和状态。
**AbortPolicy**
:处理程序遭到拒绝,则直接抛出运行时异常RejectedExecutionException。(默认策略)**CallerRunsPolicy**
:调用者所在线程来运行该任务,main 线程利用同步调用执行任务,此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。**DiscardPolicy**
:无法执行的任务将被删除。**DiscardOldestPolicy**
:如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重新尝试执行任务(如果再次失败,则重复此过程)。
工作队列BlockingQueue详解
实现BlockingQueue接口的常见类如下:
- ArrayBlockingQueue:基于数组的有界阻塞队列。队列按FIFO原则对元素进行排序,队列头部是在队列中存活时间最长的元素,队尾则是存在时间最短的元素。新元素插入到队列的尾部,队列获取操作则是从队列头部开始获得元素。 这是一个典型的“有界缓存区”,固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区,就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞;试图从空队列中提取元素将导致类似阻塞。ArrayBlockingQueue构造方法可通过设置fairness参数来选择是否采用公平策略,公平性通常会降低吞吐量,但也减少了可变性和避免了“不平衡性”,可根据情况来决策。
- LinkedBlockingQueue:基于链表的无界阻塞队列。与ArrayBlockingQueue一样采用FIFO原则对元素进行排序。基于链表的队列吞吐量通常要高于基于数组的队列。
- SynchronousQueue:同步的阻塞队列。其中每个插入操作必须等待另一个线程的对应移除操作,等待过程一直处于阻塞状态,同理,每一个移除操作必须等到另一个线程的对应插入操作。SynchronousQueue没有任何容量。不能在同步队列上进行 peek,因为仅在试图要移除元素时,该元素才存在;除非另一个线程试图移除某个元素,否则也不能(使用任何方法)插入元素;也不能迭代队列,因为其中没有元素可用于迭代。Executors.newCachedThreadPool使用了该队列。
- PriorityBlockingQueue:基于优先级的无界阻塞队列。优先级队列的元素按照其自然顺序进行排序,或者根据构造队列时提供的 Comparator 进行排序,具体取决于所使用的构造方法。优先级队列不允许使用 null 元素。依靠自然顺序的优先级队列还不允许插入不可比较的对象(这样做可能导致 ClassCastException)。虽然此队列逻辑上是无界的,但是资源被耗尽时试图执行 add 操作也将失败(导致 OutOfMemoryError)。
测试demo:
java">import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class ThreadPoolDemo {public static final ThreadPoolExecutor THREAD_POOL_EXECUTOR =new ThreadPoolExecutor(1,1,1,TimeUnit.SECONDS,new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.CallerRunsPolicy());public static void main(String[] args) {THREAD_POOL_EXECUTOR.submit(()->{try {System.out.println("第一个任务正在执行...");Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}finally{System.out.println("第一个任务执行完成...");}});System.out.println("已提交第一个任务..");THREAD_POOL_EXECUTOR.submit(()->{try {System.out.println("第二个任务正在执行...");Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}finally{System.out.println("第二个任务执行完成...");}});System.out.println("已提交第二个任务..");System.out.println("线程池资源即将耗尽..");long startTime = System.currentTimeMillis();THREAD_POOL_EXECUTOR.submit(()->{try {System.out.println("第三个任务正在执行...");Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}finally{System.out.println("第三个任务执行完成...");}});System.out.println("触发拒绝策略,共计耗时:" + (System.currentTimeMillis() - startTime));}
}