线程池详解
前提:线程池创建有两种方式,一种是Executors使用默认方法创建,另一种是通过ThreadPoolExecutor自定义,不推荐前者是因为前者的配置很多都是取得integer得最大值,很容易造成OOM
1、线程池核心概念:
int corePoolSize 核心线程数
int maximumPoolSize 最大线程数
long keepAliveTime 核心线程数满了之后创建的最大线程多久后释放
TimeUnit unit 释放时间的单位,m,h,d
BlockingQueue<Runnable> workQueue 阻塞消息队列
ThreadFactory threadFactory 线程工厂
RejectedExecutionHandler handler 拒绝策略
2、详细解释
2.1 核心线程数
当线程是IO密集型时,主要消耗磁盘的读写性能,可以设置为2*n,n为当前服务器核数(比如8核16G的服务器设置为16,Runtime.getRuntime().availableProcessors()获取)
当线程是CPU密集型时,主要消耗cpu性能,设置为n+1
2.2最大线程数
当核心线程核消息队列都满了之后才会去创建最大线程,直到达到最大线程数,之后的线程就会执行拒绝策略
2.3 阻塞消息队列
ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小,读写用一把锁,性能较差;
LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE;一般是用这个,指定了大小则限制具体大小,写核读分两把锁进行操作,所以性能较好
synchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务。
注意:当核心线程数满了之后,新线程会先存储在消息队列中,当消息队列也满了之后才会去创建最大线程,直到达到最大线程数,之后的线程就会执行拒绝策略
2.4 线程工厂
创建线程的类,可以用默认工厂,也可以自定义线程工厂实现 implements ThreadFactory类,实现newThread方法,自定义工厂的话可以设置线程名或者定义辅助线程
2.5拒绝策略
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy:丢弃任务,但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新提交被拒绝的任务
ThreadPoolExecutor.CallerRunsPolicy:由调用线程(提交任务的线程)处理该任务
自定义拒绝策略
需要继承implements RejectedExecutionHandler 实现public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)方法,此方法中可以通过类型转换的形式拿到线程具体的类,从而获取线程相关信息,详细代码见下方详细代码,需要注意的是,只能转换线程池execute的Runnable类,如果线程池执行的线程是通过实现Callable完成的,在执行前会把线程封装成FutureTask,这样相当于转换再转换,就没法转换成原来的对象了。
备注:execute只能提交Runnable线程,submit可以提交所有的Runnable、Callable、Thread线程
-------------------------------分割线-------------------------------------------------------------
详细实现代码如下
import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class ExecutorTest { public static class MyThread implements Runnable{ private Integer temp; public MyThread(Integer temp){ this.temp=temp; } @Override public void run() { System.out.println(Thread.currentThread().getName()+":MyThread--------------"+temp); try { Thread.sleep(); } catch (InterruptedException e) { e.printStackTrace(); } } } public static class MyThread2 extends Thread{ @Override public void run() { } } public static class MyThread3 implements Callable<Integer>{ Integer temp; public MyThread3(Integer temp){ this.temp=temp; } @Override public Integer call() throws Exception { System.out.println(Thread.currentThread().getName()+":MyThread3--------------"+temp); Thread.sleep(3000); return temp; } } public static void main(String[] args) { ExecutorService pools = new ThreadPoolExecutor(3, 10, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(5),new Factory(),new Handler()); try { for (int i = 0; i < 50; i++) { pools.execute(new MyThread(i)); } } finally { pools.shutdown(); } } private static class Factory implements ThreadFactory{ private AtomicInteger count = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); String threadName = "Factory----------------" + count.addAndGet(1); t.setName(threadName); System.out.println("线程" + threadName+"创建完成"); return t; } } private static class Handler implements RejectedExecutionHandler{ @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { MyThread myThread3 = (MyThread) r; System.out.println(myThread3.temp+"被阻塞了"); // r.run(); } } }