JDK线程池的使用

╰半夏微凉° 2022-08-24 11:18 201阅读 0赞

线程池代码：  
\--------------------------------------------

package cn.gaialine.threadpool;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 线程池测试用例
     * @author yangyong
     *
     */
    public class TestThreadPool {
    	//线程池维护线程的最少数量，可以根据消耗内存大小调整
    	private static final int COREPOOLSIZE = 2;
    	//线程池维护线程的最大数量，可以根据消耗内存大小调整
    	private static final int MAXINUMPOOLSIZE = 5;
    	//线程池维护线程所允许的空闲时间
    	private static final long KEEPALIVETIME = 4;
    	//线程池维护线程所允许的空闲时间的单位
    	private static final TimeUnit UNIT = TimeUnit.SECONDS;
    	//线程池所使用的缓冲队列,这里队列大小为3
    	private static final BlockingQueue<Runnable> WORKQUEUE = new ArrayBlockingQueue<Runnable>(3);
    	//线程池对拒绝任务的处理策略：AbortPolicy为抛出异常；CallerRunsPolicy为重试添加当前的任务，他会自动重复调用execute()方法；DiscardOldestPolicy为抛弃旧的任务，DiscardPolicy为抛弃当前的任务
    	//使用时注意调整这里
    	private static final AbortPolicy HANDLER = new ThreadPoolExecutor.AbortPolicy();
    
    	public static void main(String[] args) {
    		// TODO 初始化线程池
    		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(COREPOOLSIZE, MAXINUMPOOLSIZE, KEEPALIVETIME, UNIT, WORKQUEUE, HANDLER);
    		//ExecutorService executorService1 = Executors.newFixedThreadPool(3);  
    
    
    
    		for (int i = 1; i < 11; i++) {
    			String task = "task@"+i;
    			System.out.println("put->"+task);
    			//一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是 Runnable类型对象的run()方法
    			//处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务
    			//设此时线程池中的数量为currentPoolSize,若currentPoolSize>corePoolSize,则创建新的线程执行被添加的任务,
    			//当corePoolSize+workQueue>currentPoolSize>=corePoolSize,新增任务被放入缓冲队列,
    			//当maximumPoolSize>currentPoolSize>=corePoolSize+workQueue,建新线程来处理被添加的任务,
    			//当currentPoolSize>=maximumPoolSize,通过 handler所指定的策略来处理新添加的任务
    			//本例中可以同时可以被处理的任务最多为maximumPoolSize+WORKQUEUE=8个，其中最多5个在线程中正在处理，3个在缓冲队列中等待被处理
    			//当currentPoolSize>corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数
    			
    			
    			threadPool.execute(new ThreadPoolTask(task));
    			//executorService1.execute(new ThreadPoolTask(task));
    
    
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		
    		threadPool.shutdown();//关闭主线程，但线程池会继续运行，直到所有任务执行完才会停止。若不调用该方法线程池会一直保持下去，以便随时添加新的任务
    		//executorService1.shutdown();
    
    	}
    }

线程任务代码：  
\---------------------------------------

package cn.gaialine.threadpool;
    
    import java.io.Serializable;
    
    /**
     * 任务task
     * @author yangyong
     *
     */
    public class ThreadPoolTask implements Runnable,Serializable{
    	private static final long serialVersionUID = -8568367025140842876L;
    
    	private Object threadPoolTaskData;
    	private static int produceTaskSleepTime = 10000;
    	
    	public ThreadPoolTask(Object threadPoolTaskData) {
    		super();
    		this.threadPoolTaskData = threadPoolTaskData;
    	}
    
    	public void run() {
    		// TODO Auto-generated method stub
    		System.out.println("start..."+threadPoolTaskData);
    		try {
    			//模拟线程正在执行任务
    			Thread.sleep(produceTaskSleepTime);
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("stop..."+threadPoolTaskData);
    		threadPoolTaskData = null;
    	}
    	
    	public Object getTask(){
    		return this.threadPoolTaskData;
    	}
    }

执行测试，每1秒添加一个任务，每个任务执行10秒，查看打印数据：  
\-------------------------------------------------------------------------------------------

put->task@1
    start...task@1
    put->task@2
    start...task@2
    put->task@3
    put->task@4
    put->task@5
    put->task@6
    start...task@6
    put->task@7
    start...task@7
    put->task@8
    start...task@8
    put->task@9
    Exception in thread "main" java.util.concurrent.RejectedExecutionException
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
    	at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
    	at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
    	at cn.gaialine.threadpool.TestThreadPool.main(TestThreadPool.java:42)
    stop...task@1
    start...task@3
    stop...task@2
    start...task@4
    stop...task@6
    start...task@5
    stop...task@7
    stop...task@8
    stop...task@3
    stop...task@4
    stop...task@5

备注：  
\-----------------------------------  
从中可以看出task1和task2依次最先执行，这时候currentPoolSize=2达到了corePoolSize，task3、task4、task5被送入缓冲队列，达到了workQueue最大值3，  
task6、task7、task8开启新的线程开始执行，  
**此时currentPoolSize=5达到了maximumPoolSize，task9、task10根据AbortPolicy策略抛出异常，不再执行task9和task10。**

10秒钟后task1、task2....依次执行完毕释放线程，开始执行队列里的task3、task4、task5，最后task3、4、5执行完毕，所有任务完成

JDK根据ThreadPoolExecutor配置好的线程池(JDK推荐使用的线程池)：  
\------------------------------------------------------------------------------------------------

// 固定工作线程数量的线程池
        ExecutorService executorService1 = Executors.newFixedThreadPool(3);
    
        // 一个可缓存的线程池
        ExecutorService executorService2 = Executors.newCachedThreadPool();
    
        // 单线程化的Executor
        ExecutorService executorService3 = Executors.newSingleThreadExecutor();
    
        // 支持定时的以及周期性的任务执行
        ExecutorService executorService4 = Executors.newScheduledThreadPool(3);