十三、多线程执行器

十三、多线程执行器

FutureTask 和 Callable

可以用来启动一个需要很长时间的计算任务

        // Callable 与 Runnable 类似，但是有返回值
        Callable<Integer> callable=()->{
            int sum=0;
            for (int i = 0; i < 1000000000; i++) {
                sum+=i;
            }
            return sum;
        };


        // FutureTask 可以将 Callable 包装成实现了 Runnable 和 Future 接口的对象
        FutureTask<Integer> futureTask = new FutureTask<>(callable);

        // 开启线程运行任务
        Thread thread=new Thread(futureTask);
        thread.start();

        
        // 获取任务的执行结果，未计算完会阻塞
        // 如果计算线程被中断，抛出InterruptedException
        futureTask.get();
        //
        try {
            // 设置最长阻塞时间，超时还没得出结果会抛出异常
            futureTask.get(10L, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            e.printStackTrace();
        }


        // 如果计算还没开始，取消该任务。如果计算处于运行中，且参数为 true ，则计算会被中断
        futureTask.cancel(true);

        futureTask.isDone();
        futureTask.isCancelled();

        System.out.println(futureTask.get());

线程池

线程创建是有消耗的，可以提前创建一些线程，需要的时候直接使用。维护这些线程的池子，也就是线程池

执行器

通过 Executors 的工厂方法获得线程池的执行器 ExecutorService

用 ExecutorService 来执行线程并得到一个 Future

用 Future 监视线程执行状况

        // 执行器 
        ExecutorService executorService = Executors.newCachedThreadPool();
        // 必要时创建新线程，空闲线程会被保留 60秒
        Executors.newCachedThreadPool();
        // 包含固定数量的线程，空闲线程会一直被保留，参数为线程数
        Executors.newFixedThreadPool(32);
        // 只有一个线程的“池”，会按顺序执行提交的任务
        Executors.newSingleThreadExecutor();
        // 用于预定执行一些任务而构建的固定线程池，支持定时及周期性的执行任务，参数为线程数
        Executors.newScheduledThreadPool(32);
        // ScheduledThreadPool 的单线程“池”版
        Executors.newSingleThreadScheduledExecutor();


        Runnable runnable=()->{};
        Callable callable=()->1;
        Future future;
        // 提交一个 Runnable 或者 Callable 对象给 Executor 并返回一个 future 来监控执行
        // 调用 future.get() 会返回 null ,runnable 没返回值
        future=executorService.submit(runnable);
        // 调用 future.get() 会在任务完成后，把第二个参数传入的对象返回
        future=executorService.submit(runnable,new Object());

        future=executorService.submit(callable);


        // 当不在需要提交任何任务的时候调用，关闭执行器，不再接受新任务，当所有任务完成后，线程池的线程死亡
        executorService.shutdown();
        // 会取消尚未开始的任务，并且试图中断正在运行的线程
        executorService.shutdownNow();

预定和周期性的执行任务

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(12);
        
        Runnable runnable=()->{};
        Callable callable=()->1;
        // 在指定延迟后执行任务 , 在 10s 后执行任务
        executorService.schedule(runnable,10, TimeUnit.SECONDS);
        executorService.schedule(callable,10, TimeUnit.SECONDS);
        // 在指定延迟后周期性的执行任务， 10s 后开始，每天都在这个时间执行任务
        executorService.scheduleAtFixedRate(runnable,10,60*60*24,TimeUnit.SECONDS);
        // 在指定延迟后，固定间隔的执行任务， 10s 后开始，执行完任务隔 1 小时后，再次执行任务
        executorService.scheduleWithFixedDelay(runnable,10,60*60*1, TimeUnit.SECONDS);

任务组

可以控制一组任务的执行

invokeAny ： 任一任务得出结果，即返回结果

ExecutorCompletionService：可以先取出有结果的任务

        ExecutorService executorService = Executors.newFixedThreadPool(20);
        Callable<Integer> callable1=()->1;
        Callable<Integer> callable2=()->2;
        //Callable<Integer> callable2=()->{while (true);};

        List<Callable<Integer>> callables=new ArrayList();
        callables.add(callable1);
        callables.add(callable2);

        // 执行集合中的所有任务，且任意一个执行完成得出结果，即将结果返回，但是其他未执行完的线程仍会继续执行
        // 可以用在比如说多线程分段查找，任何一个线程找到了就可以返回结果了
        Integer result = executorService.invokeAny(callables);
        // 可以设置超时时间，超时未得出结果会抛出 TimeOutException
        executorService.invokeAny(callables,1,TimeUnit.SECONDS);


        // 2,1,1,2,2,1,1,1,1,1,  且每次程序运行结果不尽相同
        for (int i = 0; i < 10; i++) {System.out.print(executorService.invokeAny(callables)+",");}



        // 执行集合中所有任务,并返回 Future 的集合 ,此方法会等所有任务都有结果了才返回
        List<Future<Integer>> futures = executorService.invokeAll(callables);
        // 与 invokeAny 一样也有带超时参数的版本
        executorService.invokeAll(callables,1,TimeUnit.SECONDS);



        // 构建一个能收集完成服务执行器，内部会维护一个阻塞队列 BlockingQueue，包含已完成执行的任务的future
        ExecutorCompletionService<Integer> executorCompletionService = new ExecutorCompletionService<>(executorService);
        for (Callable<Integer> callable : callables) {
            executorCompletionService.submit(callable);
        }
        System.out.println("******");
        // take 方法取出一个已经执行完的任务的 future ，如果没有完成的任务,则会阻塞（底层的 BlockingQueue.take() 阻塞了）
        executorCompletionService.take().get();
        executorCompletionService.take().get();
        // 只有两个任务结果已经全部取出，要是再取就阻塞了
        //executorCompletionService.take().get();

        // 取出一个完成的任务结果，如果没有 返回 null
        executorCompletionService.poll();
        // 没有结果的时候会等待给定时间 （就是BlockingQueue.poll）
        executorCompletionService.poll(2,TimeUnit.SECONDS);

        executorService.shutdown();

Fork-Join 框架

Jave SE 7 提供了 fork-join 框架，可以将任务分段并行执行

一个计算1-1000（不含1000）的和 ，用法如下，

public class MyForkJoin {

    public static void main(String[] args) {

        int start=1;
        int end=1000;
        // 创建线程池
        ForkJoinPool forkJoinPool=new ForkJoinPool();
        // 创建任务,
        MyTask myTask = new MyTask(start, end);
        // 执行任务
        forkJoinPool.invoke(myTask);
        // 获得任务执行结果
        System.out.println(myTask.join());

        // 一般算法对比看计算结果是否准确
        System.out.println(sum(start,end));
    }


    static  int sum(int start,int end){
        int sum=0;
        for (int i = start; i < end; i++) {
            sum+=i;
        }
        return sum;
    }
}
// 如果需要返回计算结果 则继承 RecursiveTask<T>
// 如果不需要生成结果，则继承 RecursiveAction
class MyTask extends RecursiveTask<Integer> {
    // 设置分组每组长度的界限
    static int threshold = 10;
    // 分组起始
    int start;
    // 分组结束
    int end;

    public MyTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        // 小于每组长度就不分了 直接执行计算
        if (end - start <= threshold) {
            int sum = 0;
            for (int i = start; i < end; i++) {
                sum += i;
            }
            return sum;
        } else {
            // 继续分组
            int mid = (start + end) / 2;
            MyTask first = new MyTask(start, mid);
            MyTask second = new MyTask(mid, end);
            // 执行两个子任务 , 这明明是个任务类，那么线程池是从哪里来的呢， ForkJoinWorkerThread 内有一个 pool ，通过 Thread.currentThread() 拿到线程再拿到 pool
            invokeAll(first, second);
            // 因为这个任务的逻辑是求总和，将两个子任务求和，作为本任务的结果返回
            return first.join() + second.join();
        }
    }
}

CompletableFuture

可以在任务完成后，按顺序执行一些工作的 Future

并没有明白具体怎么用，见 执行器-可完成的Future

        // public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
        CompletableFuture completableFuture = new CompletableFuture();

        // 每个方法都返回 CompletableFuture ，在 Future 完成后执行后面的操作，且这些方法是非阻塞的
        completableFuture.thenApply((t)->1);
        completableFuture.thenCompose((t)->1);
        //  completableFuture.thenXxxxx 还有很多方法

        // 可以实现一个任务完成后按顺序执行一些工作
        completableFuture.thenApply((t)->1).thenCompose((t)->1).thenAccept((t)->{});