线程池的理解和使用（附代码说明）

r囧r小猫 2023-06-21 03:58 21阅读 0赞

# 为什么要用线程池？ #

1、降低资源的消耗。降低线程创建和销毁的资源消耗；  
2、提高响应速度：线程的创建时间为T1，执行时间T2,销毁时间T3，免去T1和T3的时间  
3、提高线程的可管理性。

# 线程池模型 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]  
我们不妨动手写一个线程池,主要有以下几个核心概念，阻塞队列中待执行的任务，线程正在执行的任务，线程数。

package com.xiangxue.ch6.mypool;
    
    import java.util.Queue;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    /** * 测试写一个线程池 */
    public class LeeThreadPoll { 
    
    
        private static int WORK_NUM = 5;//池子中可以执行的线程个数,初始参数
        private static int QUENE_COUNT = 100;//阻塞队列中允许传入的任务个数，初始参数
        private WorkThread[] workThreads;//池子中等待运行的线程
        private BlockingQueue<Runnable> queue;
        ;//阻塞队列中等待运行的任务
        private int workerNum;//池子中可以执行的线程个数,
        private int queneCount;//阻塞队列中允许传入的任务个数，
    
        //池中的线程
        private class WorkThread extends Thread { 
            @Override
            public void run() { 
                Runnable r = null;
                try { 
                    while (!interrupted()) { 
                        r = queue.take();//获取需要执行的任务
                        if (r != null) { 
                            System.out.println(getId() + " ready exec :" + r);
                            r.run();
                        }
                        r = null;//为了帮助gc
                    }
                } catch (InterruptedException e) { 
                }
            }
    
            public void stopWorker() { //中止线程
                interrupt();
            }
        }
    
        //构造方法
        // 创建线程池,worker_num为线程池中工作线程的个数
        public LeeThreadPoll(int worker_num, int taskCount) { 
            if (worker_num <= 0) worker_num = WORK_NUM;
            if (taskCount <= 0) taskCount = QUENE_COUNT;
            this.workerNum = worker_num;
            queue = new ArrayBlockingQueue<>(taskCount);
            workThreads = new LeeThreadPoll.WorkThread[worker_num];
            for (int i = 0; i < worker_num; i++) { 
                workThreads[i] = new LeeThreadPoll.WorkThread();
                workThreads[i].start();//创建线程池中工作线程
            }
            Runtime.getRuntime().availableProcessors();
        }
    
        //销毁线程池
        public void destroy() { 
            // 工作线程停止工作，且置为null
            System.out.println("ready close pool.....");
            for (int i = 0; i < workerNum; i++) { //清空并终止池中已创建的线程
                workThreads[i].stopWorker();
                workThreads[i] = null;//help gc
            }
            queue.clear();// 清空任务队列
        }
    
        //往线程池中添加任务，何时执行由线程池管理器决定
        public void execute(Runnable r) { 
            try { 
                queue.put(r);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        // 覆盖toString方法，返回线程池信息：工作线程个数和已完成任务个数
        @Override
        public String toString() { 
            return "WorkThread number:" + workerNum
                    + " wait task number:" + queue.size();
        }
    
    }

但是上面的自定义线程池还是有很多的问题的，如线程数固定，容易造成资源浪费等，但是基本原理还是一样的，接下来可以学习一下jdk中的线程池

## 线程池的父类ThreadPoolExecutor ##

构造方法如下

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) { 
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);
        }

所有线程池的父类，参数含义如下：  
**int corePoolSize** ：线程池中核心线程数，线程池中线程数量小于corePoolSize ，就会创建新线程，等于 corePoolSize ，这个任务就会保存到BlockingQueue，如果调用prestartAllCoreThreads（）方法就会一次性的启动corePoolSize 个数的线程。否则只有传入任务时线程池才创建线程，大于这个数目的并发请求会存放到队列中  
**int maximumPoolSize**, 线程池允许创建的最大线程数。==如果队列满了，并且已创建的线程数小于最大线程数，==则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数就没用了  
**long keepAliveTime**, 线程空闲下来后，存活的时间，这个参数只在大于corePoolSize才有用  
**TimeUnit unit**, 存活时间的单位值  
**BlockingQueue workQueue**, 保存任务的阻塞队列  
**ThreadFactory threadFactory,** 创建线程的工厂，给新建的线程赋予名字  
**RejectedExecutionHandler handler** ：饱和策略，当队列和线程中任务都已饱和，可定义以下四种策略

AbortPolicy ：直接抛出异常，默认；
    CallerRunsPolicy：用调用者所在的线程来执行任务
    DiscardOldestPolicy：丢弃阻塞队列里最老的任务，队列里最靠前的任务
    DiscardPolicy ：当前任务直接丢弃

线程池工作机制如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]

## 提交任务 ##

调用下面两个方法  
**execute**(Runnable command) 不需要返回 值  
Future **submit**(Callable task) 需要返回值

## 关闭线程池 ##

**shutdownNow()**:设置线程池的状态，还会尝试停止正在运行或者暂停任务的线程  
\*\*shutdown()\*\*设置线程池的状态，只会中断所有没有执行任务的线程

## 常用的几种线程池 ##

获取线程池都是在Executors这个类里有定义的

### FixedThreadPool ###

创建固定线程数量的，适用于负载较重的服务器，使用了无界队列

构造方法

public static ExecutorService newFixedThreadPool(int nThreads) { 
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }

由构造方法得出：核心线程数=最大线程数，keepAliveTime为0，意味着多余的空闲线程会被立即终止。

### SingleThreadExecutor ###

创建单个线程，需要顺序保证执行任务，不会有多个线程活动，使用了无界队列

public static ExecutorService newSingleThreadExecutor() { 
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }

### CachedThreadPool ###

==会根据需要来创建新线程的，执行很多短期异步任务的程序，==使用了SynchronousQueue（这个队列不存储数据，一有数据就传递到线程池里），即corePoolSize随着任务的传递一直变大

public static ExecutorService newCachedThreadPool() { 
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,//很大很大的值
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }

### WorkStealingPool ###

newWorkStealingPool适合使用在很耗时的操作，但是newWorkStealingPool不是ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中  
**工作密取模式**的线程池

public static ExecutorService newWorkStealingPool() { 
            return new ForkJoinPool
                (Runtime.getRuntime().availableProcessors(),//cpu数
                 ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                 null, true);
        }

### ScheduledThreadPoolExecutor ###

需要定期执行周期任务，Timer不建议使用了。  
newSingleThreadScheduledExecutor：只包含一个线程，只需要单个线程执行周期任务，保证顺序的执行各个任务  
newScheduledThreadPool 可以包含多个线程的，线程执行周期任务，适度控制后台线程数量的时候  
方法说明：  
schedule：只执行一次，任务还可以延时执行  
scheduleAtFixedRate：提交固定时间间隔的任务

## 线程池的使用 ##

package com.xiangxue.ch6;
    
    import java.util.Random;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    import com.xiangxue.tools.SleepTools;
    
    /** *@author Mark老师 享学课堂 https://enjoy.ke.qq.com * *类说明：线程池的使用 */
    public class UseThreadPool { 
    	//实现一个工作线程
        static class Worker implements Runnable
        { 
            private String taskName;
            private Random r = new Random();
    
            public Worker(String taskName){ 
                this.taskName = taskName;
            }
    
            public String getName() { 
                return taskName;
            }
    
            @Override
            public void run(){ 
                System.out.println(Thread.currentThread().getName()
                		+" process the task : " + taskName);
                SleepTools.ms(r.nextInt(100)*5);
            }
        }
        //使用Callable方式实现一个有返回值的线程
        static class CallWorker implements Callable<String>{ 
        	
            private String taskName;
            private Random r = new Random();
    
            public CallWorker(String taskName){ 
                this.taskName = taskName;
            }
    
            public String getName() { 
                return taskName;
            }    	
    
    		@Override
    		public String call() throws Exception { 
                System.out.println(Thread.currentThread().getName()
                		+" process the task : " + taskName);
                return Thread.currentThread().getName()+":"+r.nextInt(100)*5;
    		}
        	
        }
    
        public static void main(String[] args) throws InterruptedException, ExecutionException
        { 
    // 自定义一个线程池
    // ExecutorService pool = new ThreadPoolExecutor(2,4,3,TimeUnit.SECONDS,
    // new ArrayBlockingQueue<Runnable>(10),
    // new ThreadPoolExecutor.DiscardOldestPolicy());
        	ExecutorService pool = Executors.newCachedThreadPool();//获得一个线程池
        	   	for(int i=0;i<6;i++) { 
        		Worker worker = new Worker("worker_"+i);
        		pool.execute(worker);//执行任务
        	}
        	for(int i=0;i<6;i++) { 
        		CallWorker callWorker = new CallWorker("callWorker_"+i);
        		Future<String> result = pool.submit(callWorker);//submit方法是有返回值
        		System.out.println(result.get());
        	}
        	pool.shutdown();//关闭线程池
        }
    }

**Executors.newCachedThreadPool()**

对于上面有返回结果的线程，为了有序的获取线程池的线程返回结果，引入了**CompletionService**，

// 创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
            CompletionService<Integer> completionService = new ExecutorCompletionService(pool);//入参就是线程池
            // 向里面扔任务
            for (int i = 0; i < TOTAL_TASK; i++) { 
                Future<Integer> future = completionService.submit(new WorkTask("ExecTask" + i));
    
            }
    // 检查线程池任务执行结果
            for (int i = 0; i < TOTAL_TASK; i++) { 
                int sleptTime = completionService.take().get();//获取返回值
                System.out.println(" slept "+sleptTime+" ms ...");
                count.addAndGet(sleptTime);
            }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191211163949921.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191212164216460.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw,size_16,color_FFFFFF,t_70