2，Executor线程池

Posted 2020-10-29 烟丶雨下整晚

一，Executor框架简介

在Java 5之后，并发编程引入了一堆新的启动、调度和管理线程的API。Executor框架便是Java 5中引入的，其内部使用了线程池机制，它在java.util.cocurrent 包下，通过该框架来控制线程的启动、执行和关闭，可以简化并发编程的操作。

Executor框架包括：线程池，Executor，Executors，ExecutorService，CompletionService(异步任务)，Future，Callable等。

二，Executor接口

Executor接口是Executor框架中最基础的部分，定义了一个用于执行Runnable的execute方法，它没有实现类只有另一个重要的子接口ExecutorService。

public interface Executor {
    void execute(Runnable command);
}

三，ExecutorService接口

ExecutorService接口继承自Executor接口，定义了终止、提交,执行任务、跟踪任务返回结果等方法。

public interface ExecutorService extends Executor {
    //关闭线程池,当此方法被调用时，ExecutorService停止接收新的任务并且等待已经提交的任务（包含提交正在执行和提交未执行）执行完成。
    //当所有提交任务执行完毕，线程池即被关闭。
    void shutdown();
    //立即停止，将暂停所有等待处理的任务并返回这些任务的列表
    List<Runnable> shutdownNow();
    //判断执行器是否已经关闭
    boolean isShutdown();
    //判断关闭后所有任务是否都已完成 
    boolean isTerminated();
 
    //接收timeout和TimeUnit(日期工具类)两个参数，用于设定超时时间及单位。
    //当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。
    //一般情况下会和shutdown方法组合使用。
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    //提交一个Callable任务
    <T> Future<T> submit(Callable<T> task);
 
    //提交一个Runable任务，result是要返回的结果
    <T> Future<T> submit(Runnable task, T result);
 
    //提交一个任务 
    Future<?> submit(Runnable task);
 
    //执行所有给定的任务，当所有任务完成，返回保持任务状态和结果的Future列表 
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
 
    //执行给定的任务，当所有任务完成或超时期满时（无论哪个首先发生），返回保持任务状态和结果的 Future 列表。 
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;
 
    //执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
    
    //执行给定的任务，如果在给定的超时期满前某个任务已成功完成（也就是未抛出异常），则返回其结果。 
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

四，Executors类

可以Executors类来创建的线程池的类型，可以创建5种类型的线程池。

NewFixedThreadPool(numberOfThreads:int)（固定线程池）	ExecutorService 创建一个固定线程数量的线程池，并行执行的线程数量不变，线程当前任务完成后，可以被重用执行另一个任务。
NewCachedThreadPool():（可缓存线程池）	ExecutorService 创建一个线程池，按需创建新线程，就是有任务时才创建，空闲线程保存60s，当前面创建的线程可用时，则重用它们。
newSingleThreadExecutor();（单线程执行器）	线程池中只有一个线程，依次执行任务。这个线程池可以在线程死后（或发生异常时）重新启动一个线程来替代原来的线程继续执行下去。
newScheduledThreadPool()：	线程池按时间计划来执行任务，允许用户设定执行任务的时间来执行。
newSingleThreadExecutor();	线程池中只有一个线程，它按规定顺序(FIFO, LIFO, 优先级)来执行任务。

4.1，newFixedThreadPool类型

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

public static void main(String[] args) {
    ExecutorService threadPool = Executors.newFixedThreadPool(3);
    for (int i = 0; i < 12; i++) {
        int index = i;
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(index);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    //关闭线程池
    threadPool.shutdown();
}

因为线程池大小为3，每个任务输出后间隔2秒，所以每两秒打印3个数字。

4.2，newCachedThreadPool类型

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

public static void main(String[] args) {
    ExecutorService threadPool = Executors.newCachedThreadPool();
    for (int i = 0; i < 4; i++) {
        int index = i;
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("threadName="+ Thread.currentThread().getName() + " index=" + index);
            }
        });
    }
    //关闭线程池
    threadPool.shutdown();
}

线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

4.3，newSingleThreadExecutor类型

创建一个线程池（这个线程池只有一个线程）,这个线程池可以在线程死后（或发生异常时）重新启动一个线程来替代原来的线程继续执行下去。

public class Thread4 extends Thread{
    @Override
    public void run() {
        int temp = 0;  
        int i = 0;  
        Random random =new Random();  
        while(true){  
            int j =random.nextInt(100);
            System.err.println("threadName="+ Thread.currentThread().getName() + "  temp="+ temp +"  i="+ (i++) + "  j=" + j);  
            try{  
                if(temp == 0 && j > 60 ) {
                    temp = 7/0;
                }  
                Thread.sleep(100);
            }catch(Exception e){  
                e.printStackTrace();  
                temp = 1;  
            }  
        }
    }
    
    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newSingleThreadExecutor();
        Thread4 thread4 = new Thread4();
        threadPool.execute(thread4);
    }
}

4.4，newScheduledThreadPool类型

创建一个定长线程池，支持定时及周期性任务执行。

延迟3秒执行：

public static void main(String[] args) {
    ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(5);
    threadPool.schedule(new Runnable() {
        @Override
        public void run() {
            System.out.println("delay 3 seconds");
        }
    }, 3, TimeUnit.SECONDS);
    //关闭线程池
    threadPool.shutdown();
}

延迟1秒后每3秒执行一次：

public static void main(String[] args) {
    ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(5);
    threadPool.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            System.out.println("delay 3 seconds");
        }
    },1 ,3 , TimeUnit.SECONDS);
}

   

4.5，newSingleThreadExecutor类型

public static void main(String[] args) {
    ExecutorService threadPool = Executors.newSingleThreadExecutor();
    for (int i = 0; i < 10; i++) {
        final int index = i;
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(index);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

结果依次输出，相当于顺序执行各个任务。

五，Callable接口和 Future接口

5.1，Callable接口

与Runnable接口的区别在于它接收泛型，同时它执行任务后带有返回内容

//Callable同样是任务，与Runnable接口的区别在于它接收泛型，同时它执行任务后带有返回内容  
public interface Callable<V> {  
     //call方法相当于Runnable的run方法，多了一个返回值而已
     V call() throws Exception;  
} 

Runnable对象可以转换成Callable对象，需要用到Executors类，具体方法如下。

把一个Runnable对象转换成Callable对象：

public static Callable<Object> callable(Runnbale task); 

把一个Runnable和一个待返回的结果包装成一个Callable：

public static<T> Callable<T> callable(Runnbale task,T result);

5.2，Future接口

public interface Future<V> {
    
    //尝试取消一个任务，如果取消任务成功则返回true，如果取消任务失败则返回false。
    //参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。
    boolean cancel(boolean mayInterruptIfRunning);
    
    //检测任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
    boolean isCancelled();
    
    //检测任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
    boolean isDone();
    
    //获取异步任务的执行结果（如果任务没执行完将等待） 
    V get() throws InterruptedException, ExecutionException;
    
    //获取异步任务的执行结果（有最常等待时间的限制）如果在指定时间内，还没获取到结果，就直接返回null。
    //timeout表示等待的时间，unit是它时间单位 
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。

5.3，如何使用

1. 创建 Callable 接口的实现类，并实现 call() 方法，该 call() 方法将作为线程执行体，并且有返回值。

2. 创建 Callable 实现类的实例，使用 FutureTask 类来包装 Callable 对象，该 FutureTask 对象封装了该 Callable 对象的 call() 方法的返回值。

3. 使用 FutureTask 对象作为 Thread 对象的 target 创建并启动新线程。

4. 调用 FutureTask 对象的 get() 方法来获得子线程执行结束后的返回值。

/**
 * 
 * @类名称：Thread3
 * @类描述：通过 Callable 和 Future 创建线程.
 * @创建人：Zender
 */
public class Thread3 implements Callable<Integer>{
    //该方法将作为线程执行体，并且有返回值。
    @Override
    public Integer call() throws Exception {
        int i = 100;
        System.err.println("通过 Callable 和 Future 创建线程.i=" + i);
        return i;
    }
}

TestMain.java

/**
 * 
 * @类名称：TestMain
 * @类描述：测试
 * @创建人：Zender
 */
public class TestMain {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Thread3 thread3 = new Thread3();
        FutureTask<Integer> ft = new FutureTask<Integer>(thread3);
        new Thread(ft).start();
        System.out.println("线程的返回值：" + ft.get());  
    }
}