线程池的介绍

ゞ浴缸里的玫瑰 2024-04-20 11:19 84阅读 0赞

## 线程池的介绍 ##

### 一、引入 ###

#### 一个线程完成一项任务所需时间为： ####

> 1.  创建线程时间
> 2.  线程中执行任务的时间
> 3.  销毁线程时间

### 二、为什么需要线程池 ###

> 1.  线程池技术是关注如何缩短或调整创建线程和销毁线程的时间，从而提高程序的性能。项目中可以把创建线程、销毁线程分别安排在项目的启动和结束的时间段或者一些空闲的时间段
> 2.  线程池不仅调整创建线程、销毁线程产生的时间段，而且它还显著减少了创建线程的数目，提高线程的复用率
> 3.  系统启动一个新线程的成本是比较高的，因为涉及与操作系统的交互，在这种情形下，使用线程池可以很好地提高性能，尤其是当程序中需要创建大量生存期很短暂的线程时，优先考虑使用线程池

### 三、Java提供的线程池 ###

##### 1.ExecutorService：线程池的接口 #####

##### 2.Executors：创建各种线程池的工具类 #####

##### 3.创建线程池的七大参数 #####

###### 1.int corePoolSize ------------- 核心线程数量 ######

###### 2.int maximumPoolSize ------------- 最大线程数量 ######

###### 3.long keepAliveTime ------------- 闲置时间，作用于核心线程数与最大线程数之间的线程 ######

###### 4.TimeUnit unit ------------- keepAliveTime的时间单位(可以是毫秒、秒…) ######

###### 5.BlockingQueue workQueue – --------任务队列 ######

###### 6.ThreadFactory threadFactory -------- 线程工厂 ######

###### 7.RejectedExecutionHandler handler ---- 达到了线程界限和队列容量时的处理方案(拒绝策略) ######

### 四、深入源码 ###

##### 1.ExecutorService pool = Executors.newSingleThreadExecutor(); #####

##### 2.ExecutorService pool = Executors.newFixedThreadPool(3); #####

##### 3.ExecutorService pool = Executors.newCachedThreadPool(); #####

##### 这三种线程池底层都是ThreadPoolExecutor类的对象 #####

-- 分析ThreadPoolExecutor类的构造方法源码--------------------------------
    public ThreadPoolExecutor(
    	int corePoolSize,		------------- 核心线程数量 
        int maximumPoolSize,	------------- 最大线程数量 
    	long keepAliveTime,	   ------------- 闲置时间，作用于核心线程数与最大线程数之间的线程
    	TimeUnit unit,			------------- keepAliveTime的时间单位(可以是毫秒、秒....)
    	BlockingQueue<Runnable> workQueue, -- 任务队列
    	ThreadFactory threadFactory, -------- 线程工厂
    	RejectedExecutionHandler handler ---- 达到了线程界限和队列容量时的处理方案(拒绝策略)
    ) {
        }
    执行步骤:
    	1.创建线程池后
        2.任务提交后，查看是否有核心线程：
            3.1 没有 -> 就创建核心线程 -> 执行任务 -> 执行完毕后又回到线程池中
             3.2 有 -> 查看是否有闲置核心线程：
            	4.1 有 -> 执行任务 -> 执行完毕后又回到线程池
            	4.2 没有 -> 查看当前核心线程数是否核心线程数量：
            		5.1 否 -> 就创建核心线程 -> 执行任务 -> 执行完毕后又回到线程池中
            		5.2 是 -> 查看任务列表是否装载满：
            			6.1 没有 -> 就放入列表中，等待出现闲置线程
            			6.2 装满 -> 查看是否有普通线程(核心线程数到最大线程数量之间的线程)
            				7.1 没有 -> 就创建普通线程 -> 执行任务 -> 执行完毕后又回到线程池中
            				7.2 有 -> 查看是否有闲置普通线程
            					7.1.1 有 -> 执行任务 -> 执行完毕后又回到线程池中
            					7.1.2 没有 -> 查看现在所有线程数量是否为最大线程数：
            						8.1 是 -> 执行处理方案（默认处理抛出异常）
            						8.2 否 ->就创建普通线程-> 执行任务 -> 执行完毕后又回到线程池中     			
    注：
    	1.为了更好的理解，在这里区分核心线程和普通线程，实际上区分的没这么清楚，都是线程
        2.默认的处理方案就是抛出RejectedExecutionException

-- 分析单个线程的线程池的源码 --------------------------------
    ExecutorService pool = Executors.newSingleThreadExecutor();
    new ThreadPoolExecutor(
        1, -- 核心线程数量 
        1, -- 最大线程数量 
        0L, -- 闲置时间
        TimeUnit.MILLISECONDS, -- 时间单位(毫秒)
        new LinkedBlockingQueue<Runnable>() -- 无界任务队列，可以无限添加任务
    )

-- 分析指定线程的线程池的源码 --------------------------------
    ExecutorService pool = Executors.newFixedThreadPool(3);
    new ThreadPoolExecutor(
        nThreads, -- 核心线程数量 
        nThreads, -- 最大线程数量 
        0L, -- 闲置时间
        TimeUnit.MILLISECONDS, -- 时间单位(毫秒)
        new LinkedBlockingQueue<Runnable>()-- 无界任务队列，可以无限添加任务
    )

-- 创建可缓存线程的线程池 -----------------------------------
    ExecutorService pool = Executors.newCachedThreadPool();
    new ThreadPoolExecutor(
        0, -- 核心线程数量 
        Integer.MAX_VALUE,-- 最大线程数量 
        60L, -- 闲置时间
        TimeUnit.SECONDS, -- 时间单位(秒)
        new SynchronousQueue<Runnable>() -- 直接提交队列(同步队列)：没有容量队列

##### 4.总结：核心线程 -> 任务队列 -> 普通线程 -> 拒绝策略 #####

### 五、任务队列详解 ###

#### 1.LinkedBlockingQueue无界任务队列 ####

> 1.使用无界任务队列，线程池的任务队列可以无限制的添加新的任务，而线程池创建的最大线程数量就是你corePoolSize设置的数量，也就是说在这种情况下maximumPoolSize这个参数是无效的。
> 
> 2.哪怕你的任务队列中缓存了很多未执行的任务，当线程池的线程数达到corePoolSize后，就不会再增加了。
> 
> 3.若后续有新的任务加入，则直接进入队列等待，当使用这种任务队列模式时，一定要注意你任务提交与处理之间的协调与控制，不然会出现队列中的任务由于无法及时处理导致一直增长，直到最后资源耗尽的问题。

#### 2.SynchronousQueue同步任务队列直接提交任务队列 ####

> 1.使用直接提交任务队列，队列没有容量，每执行一个插入操作就会阻塞，需要再执行一个删除操作才会被唤醒，反之每一个删除操作也都要等待对应的插入操作。
> 
> 2.任务队列为SynchronousQueue，创建的线程数大于maximumPoolSize时，直接执行了拒绝策略抛出异常。
> 
> 3.使用SynchronousQueue队列，提交的任务不会被保存，总是会马上提交执行。如果用于执行任务的线程数量小于maximumPoolSize，则尝试创建新的线程。
> 
> 4.如果达到maximumPoolSize设置的最大值，则根据你设置的handler执行拒绝策略。因此这种方式你提交的任务不会被缓存起来，而是会被马上执行，在这种情况下，你需要对你程序的并发量有个准确的评估，才能设置合适的maximumPoolSize数量，否则很容易就会执行拒绝策略；

#### 3.ArrayBlockingQueue有界任务队列 ####

> 1.使用有界任务队列，若有新的任务需要执行时，线程池会创建新的线程，直到创建的线程数量达到corePoolSize时，则会将新的任务加入到等待队列中。
> 
> 2.若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数量达到maximumPoolSize设置的最大线程数量，若大于maximumPoolSize，则执行拒绝策略。
> 
> 3.在这种情况下，线程数量的上限与有界任务队列的状态有直接关系，如果有界队列初始容量较大或者没有达到超负荷的状态，线程数将一直维持在corePoolSize以下，反之当任务队列已满时，则会以maximumPoolSize为最大线程数上限。

#### 4.PriorityBlockingQueue优先任务队列 ####

> 使用优先任务队列，它其实是一个特殊的无界队列，它其中无论添加了多少个任务，线程池创建的线程数也不会超过corePoolSize的数量，只不过其他队列一般是按照先进先出的规则处理任务，而PriorityBlockingQueue队列可以自定义规则根据任务的优先级顺序先后执行。

##### 优先队列的使用说明： #####

public class Test {
        
    	
    	public static void main(String[] args) {
        
    		
    		ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 200, 1000, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
    		
    		for (int i = 1; i <= 10; i++) {
        
    			pool.execute(new Task(i));
    		}
    		
    		pool.shutdown();
    		
    	}
    }

public class Task implements Runnable,Comparable<Task>{
        
    	
    	private int priority;
    	
    	public Task(int priority) {
        
    		this.priority = priority;
    	}
    
    	@Override
    	public void run() {
        
    		try {
        
    			Thread.sleep(1000);
    			System.out.println(Thread.currentThread().getName() + " priority：" + this.priority);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    	}
    
    	//当前对象和其他对象做比较，当前优先级大就返回-1，优先级小就返回1,值越小优先级越高
    	@Override
    	public int compareTo(Task o) {
        
    		return (this.priority>o.priority)?-1:1;
    	}
    }
    总结：除了第一个任务直接创建线程执行外，其他的任务都被放入了优先任务队列，按优先级进行了重新排列执行，且线程池的线程数一直为corePoolSize，也就是只有一个。

### 六、拒绝策略 ###

#### ThreadPoolExecutor自带的拒绝策略有四种，都实现了RejectedExecutionHandler接口 ####

> ##### 比如：new ThreadPoolExecutor.AbortPolicy() #####

#### 1.AbortPolicy ####

> 1.当有任务添加到线程池被拒绝时，会抛出RejectedExecutionException异常，线程池默认的拒绝策略。
> 
> 2.必须处理好抛出的异常，否则会打断当前的执行流程，影响后续的任务执行。

#### 2.DiscardPolicy ####

> 当有任务添加到线程池被拒绝时，直接丢弃，其他啥都没有

#### 3.CallerRunsPolicy ####

> 1.当有任务添加到线程池被拒绝时，线程池会将被拒绝的任务添加到线程池正在运行的线程中去运行。
> 
> 2.一般并发比较小，性能要求不高，不允许失败。但是，由于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能效率上必然的损失较大

#### 4.DiscardOledestPolicy ####

> 当有任务添加到线程池被拒绝时，线程池会丢弃阻塞队列中末尾的任务（最老的任务–第一个添加的任务），然后将被拒绝的任务添加到末尾。 如果项目中有允许丢失任务的需求，可以使用

### 七、自定义拒绝策略 ###

public class Test {
        
    	
    	public static void main(String[] args) {
        
    		
    		ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1), Executors.defaultThreadFactory(), new RejectedExecutionHandler() {
        
    			@Override
    			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
    				System.out.println(r.toString()+"执行了拒绝策略");
    			}
    		});
    		
    		for (int i = 1; i <= 10; i++) {
        
    			pool.execute(new Task());
    		}
    		
    		pool.shutdown();
    	}
    }

public class Task implements Runnable{
        
    	
    	@Override
    	public void run() {
        
    		try {
        
    			Thread.sleep(1000);
    			System.out.println(Thread.currentThread().getName());
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    	}
    }

### 八、自定义线程池原因 ###

> 1.在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供，不允许在应用中自行显示的创建线程，这样一方面使线程的创建更加规范，可以合理控制开辟线程的数量；另一方面线程的细节管理交给线程池处理，优化了资源的开销。而线程池不允许使用Executors去创建，而要通过ThreadPoolExecutor方式，这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()等创建线程池的方法，但都有其局限性，不够灵活，而且有资源耗尽的风险（OOM - Out Of Memory ）。
> 
> 2.一般我们创建线程池时，为防止资源被耗尽，任务队列都会选择创建有界任务队列，但这种模式下如果出现任务队列已满且线程池创建的线程数达到你设置的最大线程数时，这时就需要你指定ThreadPoolExecutor的RejectedExecutionHandler参数即合理的拒绝策略，来处理线程池"超载"的情况

### 九、自定义线程池 ###

> 线程池中线程就是通过ThreadPoolExecutor中的ThreadFactory–线程工厂创建的。那么通过自定义ThreadFactory，可以按需要对线程池中创建的线程进行一些特殊的设置，如命名、优先级等

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Test01 {
    
    	public static void main(String[] args) {
    		
    		ThreadPoolExecutor pool = new ThreadPoolExecutor(
    				5, //核心线程
    				20,//最大线程数
    				60L,//闲置时间
    				TimeUnit.SECONDS,//时间单位 
    				new ArrayBlockingQueue<>(30), //有界队列 
    				new ThreadFactory() {//定义线程工厂
    					
    					private int count;//线程编号
    					@Override
    					public Thread newThread(Runnable r) {
    						Thread thread = new Thread(r,"线程" + count);
    						count++;
    						return thread;
    					}
    				}, 
    				new RejectedExecutionHandler() {//自定义拒绝策略
    					//当任务被拒绝时调用的方法
    					@Override
    					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    						System.out.println("被拒绝的任务是：" + r);
    
    					}
    				}){
    
    			@Override
    			protected void beforeExecute(Thread t, Runnable r) {
    				System.out.println(r + "--任务处理前调用的方法");
    			}
    
    			@Override
    			protected void afterExecute(Runnable r, Throwable t) {
    				System.out.println(r + "--任务处理后调用的方法");
    			}
    
    			@Override
    			protected void terminated() {
    				System.out.println("关闭线程池时调用的方法");
    			}
    		};
    		
    		for (int i = 1; i <=100; i++) {
    			Task task = new Task(i);
    			pool.execute(task);
    		}
    		
    		pool.shutdown();
    	}
    }

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Test02 {
        
    
    	public static void main(String[] args) {
        
    		
    	 ThreadPoolExecutor pool = new FastThreadPool(
    			 5,  //核心线程
    			 20, //最大线程数
    			 60L, //闲置时间
    			 TimeUnit.SECONDS, //时间单位 
    			 new ArrayBlockingQueue<>(30), //有界队列 
    			 new RejectedExecutionHandler() {
        //自定义拒绝策略
    				//当任务被拒绝时调用的方法
    				@Override
    				public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
    				System.out.println("被拒绝的任务是：" + r);
    					
    				}
    			});
    	 for (int i =1; i <=100; i++) {
        
    		Task task = new Task(i);
    		pool.execute(task);
    		
    	}
    	 pool.shutdown();
    	}
    }

import java.time.LocalDateTime;
    
    public class Task implements Runnable{
        
    
    	//任务编号
    	private int num;
    
    	public Task(int num) {
        
    		this.num = num;
    	}
    	
    	@Override
    	public void run() {
        
    		System.out.println(Thread.currentThread().getName() + "---" + num + "--" + LocalDateTime.now());
    		
    	}
    
    	@Override
    	public String toString() {
        
    		return "任务" + num ;
    	}	
    }

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    //自定义线程池
    public class FastThreadPool extends ThreadPoolExecutor{
        
    
    	public FastThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
    			BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        
    		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    	}
    
    	public FastThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
    			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        
    		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,new MyThreadFactory(), handler);
    	}
    
    	@Override
    	protected void beforeExecute(Thread t, Runnable r) {
        
    		System.out.println(r + "-- 任务处理前调用的方法");
    	}
    	
    	@Override
    	protected void afterExecute(Runnable r, Throwable t) {
        
    		System.out.println(r + "--任务处理后的调用方法");
    	}
    	
    	@Override
    	protected void terminated() {
        
    		System.out.println("关闭线程池时调用的方法");
    	}
    	
    	//自定义线程工厂
    	private static class MyThreadFactory implements ThreadFactory{
        
    		
    		private int count;//线程编号
    		
    		@Override
    		public Thread newThread(Runnable r) {
        
    			Thread thread = new Thread(r,"线程" + count);
    			count++;
    			return thread;
    		}		
    	}	
    }

### 十、ThreadPoolExecutor扩展 ###

> ThreadPoolExecutor扩展主要是围绕beforeExecute()、afterExecute()和terminated()三个方法的重写， 通过这三个方法我们可以监控每个任务的开始和结束时间，或者其他一些功能。

#### 1.beforeExecute ####

> ##### 线程池中任务运行前执行 #####

#### 2.afterExecute ####

> ##### 线程池中任务运行完毕后执行 #####

#### 3.terminated ####

> ##### 线程池退出后执行 #####

### 十一、线程池线程数量 ###

#### 实际工作中应该使用 sysbench多线程性能测试工具 ，计算线程池线程数量，在进行编写代码。 ####

### 十二、带有返回值的任务类 ###

#### 线程池提交任务：execute()和submit() ####

> execute(Runnable)
> 
> submit(Runnable/Callable) --> 返回Future（Callable的返回值将存入Future对象）

需求：计算任务，一个包含了2万个整数的数组，分拆了多个线程来进行并行计算，最后汇总出计算的结果。

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.Future;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Test01 {
        
    
    	public static void main(String[] args) throws InterruptedException,Exception{
        
    		
    		//创建包含了2万个整数的数组
    		int[] nums = new int[20000];
    		
    		//初始化数组中的数据(1~20000)
    		for (int i = 0; i < nums.length; i++) {
        
    			nums[i] = i + 1;
    		}
    		
    		//创建线程池的对象
    		FastThreadPool pool = new FastThreadPool(4,4,0L,TimeUnit.SECONDS,new ArrayBlockingQueue<>(1),new ThreadPoolExecutor.AbortPolicy());
    		
    		//创建任务
    		Task task1 = new Task(nums,0,5000);
    		Task task2 = new Task(nums,5000,10000);
    		Task task3 = new Task(nums,10000,15000);
    		Task task4 = new Task(nums,15000,20000);
    	
    		//提交任务
    		Future<Integer> f1 = pool.submit(task1);
    		Future<Integer> f2 = pool.submit(task2);
    		Future<Integer> f3 = pool.submit(task3);
    		Future<Integer> f4 = pool.submit(task4);
    		
    		//获取任务对象的返回值
    		int result = f1.get() + f2.get() + f3.get() + f4.get();
    		System.out.println(result);
    		
    		pool.shutdown();
    	}
    }

import java.util.concurrent.Callable;
    //带有返回值的任务类
    public class Task implements Callable<Integer>{
        
    	
    	private int[] nums;
    	private int startIndex;
    	private int endIndex;
    	
    	public Task(int[] nums, int startIndex, int endIndex) {
        
    		this.nums = nums;
    		this.startIndex = startIndex;
    		this.endIndex = endIndex;
    	}
    	
    	@Override
    	public Integer call() throws Exception {
        
    		int sum = 0;
    		for (int i = startIndex; i < endIndex; i++) {
        
    			sum += nums[i];
    		}
    		return sum;
    	}
    }