并发编程预定义线程池解析（二十八）

淩亂°似流年 2022-12-31 02:28 90阅读 0赞

今天分析并发编程预定义线程池原理：

**FixedThreadPool**  **详解**

创建使用固定线程数的  FixedThreadPool  的  API 。适用于为了满足资源管理的 需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。

FixedThreadPool  的  corePoolSize  和  maximumPoolSize  都被设置为创建 FixedThreadPool 时指定的参数  nThreads 。 当线程池中的线程数大于 corePoolSize  时， keepAliveTime  为多余的空闲线程 等待新任务的 最长时间，超过这个时间后多余的线程将被终止。这里把 keepAliveTime  设 置为 0L ，意味着多余的空闲线程会被立即终止。 FixedThreadPool 使用有界队列  LinkedBlockingQueue  作为线程池的工作队列 （队列的容量为 Integer.MAX\_VALUE ）。

**SingleThreadExecutor**

创建使用单个线程的  SingleThread-Executor  的  API ，于需要保证顺序地执行 各个任务；并且在任意时间点，不会有多个线程是活动的应用场景。 corePoolSize 和  maximumPoolSize  被设置为  1 。其他参数与  FixedThreadPool 相同。SingleThreadExecutor  使用有界队列  LinkedBlockingQueue  作为线程池的工 作队列（队列的容量为 Integer.MAX\_VALUE ）。

**CachedThreadPool**

创建一个会根据需要创建新线程的  CachedThreadPool  的  API 。大小无界的线 程池，适用于执行很多的短期异步任务的小程序，或者是负载较轻的服务器。 corePoolSize 被设置为  0 ，即  corePool  为空； maximumPoolSize  被设置为 Integer.MAX\_VALUE。这里把  keepAliveTime  设置为  60L ，意味着  CachedThreadPool 中的空闲线程等待新任务的最长时间为60 秒，空闲线程超过 60 秒后将会被终止。 FixedThreadPool 和  SingleThreadExecutor  使用有界队列  LinkedBlockingQueue 作为线程池的工作队列。CachedThreadPool  使用没有容量的  SynchronousQueue 作为线程池的工作队列，但 CachedThreadPool  的  maximumPool  是无界的。这意味着，如果主线程提交任务的速度高于 maximumPool  中线程处理任务的速度时，

CachedThreadPool  会不断创建新线程。极端情况下， CachedThreadPool  会因为创 建过多线程而耗尽 CPU  和内存资源。

**WorkStealingPool**

利用所有运行的处理器数目来创建一个工作窃取的线程池，使用  forkjoin  实 现

**ScheduledThreadPoolExecutor**

使用工厂类  Executors  来创建。 Executors  可以创建  2  种类型的 ScheduledThreadPoolExecutor，如下。

• ScheduledThreadPoolExecutor 。包含若干个线程的 ScheduledThreadPoolExecutor。

• SingleThreadScheduledExecutor 。只包含一个线程的 ScheduledThreadPoolExecutor。

ScheduledThreadPoolExecutor  适用于需要多个后台线程执行周期任务，同时 为了满足资源管理的需求而需要限制后台线程的数量的应用场景。

SingleThreadScheduledExecutor  适用于需要单个后台线程执行周期任务，同 时需要保证顺序地执行各个任务的应用场景。

***提交定时任务***

public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)

// 向定时任务线程池提交一个延时  Runnable  任务（仅执行一次）

public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);

// 向定时任务线程池提交一个延时的  Callable  任务（仅执行一次）

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

// 向定时任务线程池提交一个固定时间间隔执行的任务

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);

// 向定时任务线程池提交一个固定延时间隔执行的任务

固定时间间隔的任务不论每次任务花费多少时间，下次任务开始执行时间从 理论上讲是确定的，当然执行任务的时间不能超过执行周期。 固定延时间隔的任务是指每次执行完任务以后都延时一个固定的时间。由于 操作系统调度以及每次任务执行的语句可能不同，所以每次任务执行所花费的时 间是不确定的，也就导致了每次任务的执行周期存在一定的波动。

测试代码：

/**
     *类说明：演示ScheduledThreadPoolExecutor的用法
     */
    public class ScheduledCase {
        public static void main(String[] args) {
        	
        	ScheduledThreadPoolExecutor schedule
                    = new ScheduledThreadPoolExecutor(1);
        	
            //延时Runnable任务（仅执行一次）
            schedule.schedule(new Runnable() {
                @Override
                    public void run() {
                    System.out.println("the task only run once!");
                }
            },3000,TimeUnit.MILLISECONDS);
            //固定延时间隔执行的任务
            schedule.scheduleWithFixedDelay(new Runnable() {
                @Override
                public void run() {
                    System.out.println("*******fixDelay start,"
                            +ScheduleWorker.formater.format(new Date()));
                    SleepTools.second(2);
                    System.out.println("*******fixDelay end,"
                            +ScheduleWorker.formater.format(new Date()));
                }
            },1000,3000,TimeUnit.MILLISECONDS);
    
        	
            //固定时间间隔执行的任务,从理论上说第二次任务在6000 ms后执行，第三次在
            //6000*2 ms后执行
            schedule.scheduleAtFixedRate(
                    new ScheduleWorkerTime(),0,6000,
                    TimeUnit.MILLISECONDS);
    
    //        // 固定时间间隔执行的任务,开始执行后就触发异常,next周期将不会运行
            schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.HasException),
                    0, 3000, TimeUnit.MILLISECONDS);
    
            // 固定时间间隔执行的任务,虽然抛出了异常,但被捕捉了,next周期继续运行
            schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.ProcessException),
                    0, 3000, TimeUnit.MILLISECONDS);
            
        }
    }

辅助代码：

/**
     *类说明：定时任务的工作类
     */
    public class ScheduleWorker implements Runnable{
        public final static int Normal = 0;//普通任务类型
        public final static int HasException = -1;//会抛出异常的任务类型
        public final static int ProcessException = 1;//抛出异常但会捕捉的任务类型
    
        public static SimpleDateFormat formater = new SimpleDateFormat(
                "yyyy-MM-dd HH:mm:ss");
        
        private int taskType;
        public ScheduleWorker(int taskType) {
            this.taskType = taskType;
        }
    
        @Override
        public void run() {
        	if(taskType==HasException) {
                System.out.println(formater.format(new Date())
                        +" Exception be made,will next task run?");
                throw new RuntimeException("ExceptionHappen");    		
        	}else if(taskType==ProcessException) {
                try {
                    System.out.println("ProcessException ..."
                    		+formater.format(new Date()));
                    throw new RuntimeException("ProcessException");
                } catch (RuntimeException e) {
                    System.out.println("ProcessException catched,,will next task run?");
                }  		
        	}else {
        		System.out.println("Normal..."+formater.format(new Date()));
        	}
        }
    }

***解释定时任务超时问题***

scheduleAtFixedRate  中，若任务处理时长超出设置的定时频率时长，本次任 务执行完才开始下次任务，下次任务已经处于超时状态，会马上开始执行。 若任务处理时长小于定时频率时长，任务执行完后，定时器等待，下次任务 会在定时器等待频率时长后执行。

如下例子：

设置定时任务每  60s  执行一次，那么从理论上应该第一次任务在第  0s  开始 ,

第二次任务在第  60s  开始，第三次任务在  120s  开始，但实际运行时第一次任务 时长 80s ，第二次任务时长  30s ，第三次任务时长  50s ，则实际运行结果为：

第一次任务第  0s  开始 , 第  80s  结束；

第二次任务第  80s  开始 , 第  110s  结束 ( 上次任务已超时 , 本次不会再等待  60s, 会马上开始) ；

第三次任务第  120s  开始 , 第  170s  结束 .

第四次任务第  180s  开始 .....

测试代码

/**
     *类说明：定时任务的工作类
     */
    public class ScheduleWorkerTime implements Runnable{
        public final static int Long_8 = 8;//工作8秒
        public final static int Short_2 = 2;//工作2秒
        public final static int Normal_5 = 5;//工作5秒
    
        public static SimpleDateFormat formater = new SimpleDateFormat(
                "yyyy-MM-dd HH:mm:ss");
        public static AtomicInteger count = new AtomicInteger(0);
        
        @Override
        public void run() {
        	if(count.get()==0) {
                System.out.println("Long_8....begin:"+formater.format(new Date()));
                SleepTools.second(Long_8);
                System.out.println("Long_8....end:"+formater.format(new Date())); 
                count.incrementAndGet();
        	}else if(count.get()==1) {
        		System.out.println("Short_2 ...begin:"+formater.format(new Date()));
        		SleepTools.second(Short_2);
        		System.out.println("Short_2 ...end:"+formater.format(new Date()));
                count.incrementAndGet();    		
        	}else {
        		System.out.println("Normal_5...begin:"+formater.format(new Date()));
        		SleepTools.second(Normal_5);
        		System.out.println("Normal_5...end:"+formater.format(new Date()));
        		count.incrementAndGet(); 
        	}
    //    	if(taskType==Long_8) {
    //            System.out.println("Long_8....begin:"+formater.format(new Date()));
    //            SleepTools.second(Long_8);
    //            System.out.println("Long_8....end:"+formater.format(new Date()));
    //    	}else if(taskType==Short_2) {
    //    		System.out.println("Short_2 ...begin:"+formater.format(new Date()));
    //    		SleepTools.second(Short_2);
    //    		System.out.println("Short_2 ...end:"+formater.format(new Date()));
    //    	}else {
    //    		System.out.println("Normal_5...begin:"+formater.format(new Date()));
    //    		SleepTools.second(Normal_5);
    //    		System.out.println("Normal_5...end:"+formater.format(new Date()));
    //    	}
        }
    }

**CompletionService**

CompletionService  实际上可以看做是  Executor  和  BlockingQueue  的结合体。

CompletionService  在接收到要执行的任务时，通过类似  BlockingQueue  的  put  和 take 获得任务执行的结果。

CompletionService  的一个实现是  ExecutorCompletionService ， ExecutorCompletionService 把具体的计算任务交给  Executor  完成。 在实现上，ExecutorCompletionService  在构造函数中会创建一个 BlockingQueue（使用的基于链表的  LinkedBlockingQueue ），该  BlockingQueue  的 作用是保存 Executor  执行的结果。 当提交一个任务到 ExecutorCompletionService  时，首先将任务包装成 QueueingFuture，它是  FutureTask  的一个子类，然后改写  FutureTask  的  done  方 法，之后把 Executor  执行的计算结果放入  BlockingQueue  中。 与 ExecutorService  最主要的区别在于  submit  的  task  不一定是按照加入时的 顺序完成的。CompletionService  对  ExecutorService  进行了包装，内部维护一个保 存 Future  对象的  BlockingQueue 。只有当这个  Future  对象状态是结束的时候，才 会加入到这个 Queue  中，take()方法其实就是  Producer-Consumer  中的  Consumer 。它会从 Queue  中取出  Future  对象，如果  Queue  是空的，就会阻塞在那里，直到 有完成的 Future  对象加入到  Queue  中。所以，先完成的必定先被取出。这样就 减少了不必要的等待时间。

核心代码：

public class CompletionCase {
        private final int POOL_SIZE = 3;
    //    private final int POOL_SIZE = Runtime.getRuntime().availableProcessors();
        private final int TOTAL_TASK = 5;
    //    private final int TOTAL_TASK = Runtime.getRuntime().availableProcessors()*10;
    
        // 方法一，自己写集合来实现获取线程池中任务的返回结果
        public void testByQueue() throws Exception {
        	long start = System.currentTimeMillis();
        	AtomicInteger count = new AtomicInteger(0);
            // 创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
            //队列,拿任务的执行结果
            BlockingQueue<Future<Integer>> queue = 
            		new LinkedBlockingQueue<>();
    
            // 向里面扔任务
            for (int i = 0; i < TOTAL_TASK; i++) {
                Future<Integer> future = pool.submit(new WorkTask("ExecTask" + i));
                queue.add(future);
            }
    
            // 检查线程池任务执行结果
            for (int i = 0; i < TOTAL_TASK; i++) {
            	int sleptTime = queue.take().get();
            	System.out.println(" slept "+sleptTime+" ms ...");
            	count.addAndGet(sleptTime);
            }
    
            // 关闭线程池
            pool.shutdown();
            System.out.println("-------------tasks sleep time "+count.get()
            		+"ms,and spend time "
            		+(System.currentTimeMillis()-start)+" ms");
        }
    
        public void testByCompletion() throws Exception{
            long start = System.currentTimeMillis();
            AtomicInteger count = new AtomicInteger(0);
            // 创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
            CompletionService<Integer> cSevice
                    = new ExecutorCompletionService<>(pool);
    
            // 向里面扔任务
            for (int i = 0; i < TOTAL_TASK; i++) {
                cSevice.submit(new WorkTask("ExecTask" + i));
            }
    
            // 检查线程池任务执行结果
            for (int i = 0; i < TOTAL_TASK; i++) {
                int sleptTime = cSevice.take().get();
                System.out.println(" slept结果 "+sleptTime+" ms ...");
                count.addAndGet(sleptTime);
            }
    
            // 关闭线程池
            pool.shutdown();
            System.out.println("-------------tasks sleep time "+count.get()
                    +"ms,and spend time "
                    +(System.currentTimeMillis()-start)+" ms");
    
        }
    
        public static void main(String[] args) throws Exception {
            CompletionCase t = new CompletionCase();
            t.testByQueue();
            t.testByCompletion();
        }
    }

任务线程：

public class WorkTask implements Callable<Integer> {
        private String name;
        public WorkTask(String name) {
            this.name = name;
        }
    
        @Override
        public Integer call() {
            int sleepTime = new Random().nextInt(1000);
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            // 返回给调用者的值
            return sleepTime;
        }
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

我们可以得出结论：

使用方法一，自己创建一个集合来保存 Future  存根并循环调用其返回结果 的时候，主线程并不能保证首先获得的是最先完成任务的线程返回值。它只是按 加入线程池的顺序返回。因为 take  方法是阻塞方法，后面的任务完成了，前面的任务却没有完成，主程序就那样等待在那儿，只到前面的完成了，它才知道原 来后面的也完成了。

使用方法二，使用 CompletionService  来维护处理线程不的返回结果时，主 线程总是能够拿到最先完成的任务的返回值，而不管它们加入线程池的顺序。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/c3d62bfef0fb48e0b9c1d8cb32a4233b.png