并发编程(14)-线程池

小鱼儿 2022-06-05 03:06 326阅读 0赞

# 1.Executor #

为了更好的控制多线程，JDK提供了一套线程框架Executor，帮助开发人员有效的进行线程控制。他们都在java.util.concurrent包中，是JDK并发包的核心。其中有一个比较重要的类：Executors，他扮演者线程工厂的角色，我们通过Executors可以创建特定功能的线程池。  
Executors创建线程的方法：  
**newFixedThreadPool()**方法，该方法返回一个固定数量的线程池，该方法的线程数始终不变，当有一个任务提交时，若线程池中空闲，则立即执行，若没有，则会被暂缓则一个任务队列中等待有空闲的线程去执行。  
**newSingleThreadExecutor()**方法，创建一个线程的线程池，若空闲则执行，若没有空闲线程则暂缓在队列任务中。  
**newCachedThreadPool()**方法，返回一个可根据实际情况调整线程个数的线程池，不限制最大线程数量，若有空闲的线程则执行任务，若无任务则不创建线程，并且每一个空闲线程会在60秒后自动回收。  
**newScheduledThreadPool()**方法，该方法返回一个SchededExecutorService对象，但该线程池可以指定线程的数量。

1.newfixedThreadPool()

ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);

进入到newFixedThreadPool

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

2.newSingleThreadExecutor()方法

ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();

进入到newSingleThreadExecutor底层

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

3.newCachedThreadPool()方法

ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();

进入到newCachedThreadPool()底层

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

4.newScheduledThreadPool()方法

ExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(10);

进入到newScheduledThreadPool()底层

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
            return new ScheduledThreadPoolExecutor(corePoolSize);
        }
    
    public ScheduledThreadPoolExecutor(int corePoolSize) {
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }
    //进入到super
    public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }

从上面提供的四个方法可以看到都是通过ThreadPoolExecutor实例对象。

# 2.自定义线程池 #

通过上面的底层代码，都是new ThreadPoolExecutor()实现，那么如果上面的四种线程池都无法满足到实际的需求，那么可以通过自己new对象，传递不同的参数实现不同的功能。但是在自定义的时候，我们需要了解该对象的各个参数表达的含义：

/** * corePoolSize 核心线程数 * maximumPoolSize 最大线程数 * keepAliveTime 线程保持活着的时间（空闲的时间） * unit 空闲时间的时间单位 * workQueue 无空闲线程时，将新加入的任务放在该队列中 * threadFactory * handler 当线程池无法再接收任务是，执行该方法拒绝执行后续的任务 */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

DEMO

package com.thread.mythread.conn019;
    
    public class MyTask implements Runnable { 
    
        private int taskId;
        private String taskName;
    
        public MyTask(int taskId, String taskName){
            this.taskId = taskId;
            this.taskName = taskName;
        }
    
        public int getTaskId() {
            return taskId;
        }
    
        public void setTaskId(int taskId) {
            this.taskId = taskId;
        }
    
        public String getTaskName() {
            return taskName;
        }
    
        public void setTaskName(String taskName) {
            this.taskName = taskName;
        }
    
        public void run() {
            try {
                System.out.println("run taskId =" + this.taskId);
                Thread.sleep(5*1000);
                //System.out.println("end taskId =" + this.taskId);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }       
        }
    
        public String toString(){
            return Integer.toString(this.taskId);
        }
    
    }

package com.thread.mythread.conn019;
    
    
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadPoolExecutor;
    
    public class MyRejected implements RejectedExecutionHandler{ 
    
    
        public MyRejected(){
        }
    
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            System.out.println("自定义处理..");
            System.out.println("当前被拒绝任务为：" + r.toString());
        }
    
    }

package com.thread.mythread.conn019;
    
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class UseThreadPoolExecutor1 { 
    
    
        public static void main(String[] args) {
            /** * 在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程， * 若大于corePoolSize，则会将任务加入队列， * 若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程， * 若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。 * */ 
            ThreadPoolExecutor pool = new ThreadPoolExecutor(
                    1,              //coreSize
                    2,              //MaxSize
                    60,             //60
                    TimeUnit.SECONDS, 
                    new ArrayBlockingQueue<Runnable>(3)         //指定一种队列 （有界队列）
                    //new LinkedBlockingQueue<Runnable>()
                    , new MyRejected()
                    //, new DiscardOldestPolicy()
                    );
    
            MyTask mt1 = new MyTask(1, "任务1");
            MyTask mt2 = new MyTask(2, "任务2");
            MyTask mt3 = new MyTask(3, "任务3");
            MyTask mt4 = new MyTask(4, "任务4");
            MyTask mt5 = new MyTask(5, "任务5");
            MyTask mt6 = new MyTask(6, "任务6");
    
            pool.execute(mt1);
            pool.execute(mt2);
            pool.execute(mt3);
            pool.execute(mt4);
            pool.execute(mt5);
            pool.execute(mt6);
    
            pool.shutdown();
    
        }
    }

package com.thread.mythread.conn019;
    
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class UseThreadPoolExecutor2 implements Runnable{ 
    
        private static AtomicInteger count = new AtomicInteger(0);
    
        public void run() {
            try {
                int temp = count.incrementAndGet();
                System.out.println("任务" + temp);
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws Exception{
            //System.out.println(Runtime.getRuntime().availableProcessors());
            BlockingQueue<Runnable> queue = 
                    //new LinkedBlockingQueue<Runnable>();
                    new ArrayBlockingQueue<Runnable>(10);
            ExecutorService executor  = new ThreadPoolExecutor(
                        5,      //core
                        10,     //max
                        120L,   //2fenzhong
                        TimeUnit.SECONDS,
                        queue);
    
            for(int i = 0 ; i < 20; i++){
                executor.execute(new UseThreadPoolExecutor2());
            }
            Thread.sleep(1000);
            System.out.println("queue size:" + queue.size());       //10
            Thread.sleep(2000);
        }
    
    
    }

在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程，若大于corePoolSize，则会将任务键入队列中，若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程，若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。  
无界的任务队列，linkedblockingqueue，与有界队列相比，除非系统资源耗尽，否则无界的任务队列不存在任务入队失败的情况。当有新任务到来，系统的线程数小于corePoolSize时，则新建线程执行任务。当达到corePoolSize后，就不会继续增加。若后续仍有新的任务加入，而没有空闲的线程资源，则任务直接进入队列等待。若任务创建和处理的速度差异很大，无界队列会保持快速增长，知道耗尽系统内存。