Java高并发编程（四）

  一、Executor执行器

　　1.Executor接口，java线程池框架中的顶层接口，提供一个execute方法来执行任务

import java.util.concurrent.Executor;

public class T01_MyExecutor implements Executor {

    public static void main(String[] args) {
        new T01_MyExecutor().execute(()->System.out.println("hello executor"));
    }

    @Override
    public void execute(Runnable command) {
        //new Thread(command).run();//另起线程方法调用
        command.run();//方法直接调用
    }

}

　　2.任务接口

　　　　1）callable接口：提供一个call方法，具有返回值可以抛出异常

　　　　2）runnable接口：提供一个run方法，无返回值不可抛出异常

　　3.ExecutorService接口（Executor的一个子接口）

　　　　1）ExecutorService可以理解为服务，执行execute方法可以向服务器中添加runnable任务无返回值，而执行submit方法可以向服务中添加callable与runnable任务，且有返回值

　　　　2）Executor与ExecutorService的简单理解：可以想象许多的执行器Executor在等待着执行任务，而service服务可通过调用execute或submit方法来添加不同的任务供执行器执行

　　　　3）submit方法具有一个future类型的返回值，与callable，Executors一起使用

public class T06_Future {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        
        FutureTask<Integer> task = new FutureTask<>(()->{
            TimeUnit.MILLISECONDS.sleep(500);
            return 1000;
        }); //new Callable () { Integer call();}
        
        new Thread(task).start();
        
        System.out.println(task.get()); //阻塞
        
        //*******************************
        ExecutorService service = Executors.newFixedThreadPool(5);
        Future<Integer> f = service.submit(()->{
            TimeUnit.MILLISECONDS.sleep(500);
            return 1;
        });
        System.out.println(f.get());
        System.out.println(f.isDone());
        
    }
}

运行结果：

　　4.Executors（操作Executor的一个工具类、工厂类）

  二、ThreadPool线程池

　　1.线程池将任务分配给池中线程，当线程池中线程执行不过来时，任务会进入等待队列，队列由BlockingQueue实现，当有线程空闲时再来任务时会被分配给空闲线程，一个线程同时维护着两个队列（结束队列与等待队列）

　　下面一个程序帮助理解线程池的概念

public class T05_ThreadPool {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(5); //execute submit
        for (int i = 0; i < 6; i++) {
            service.execute(() -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName());
            });
        }
        System.out.println(service);
        
        service.shutdown();//正常关闭，等待所有任务执行完毕，关闭线程池
        System.out.println(service.isTerminated());//判断所有任务是否都执行完了
        System.out.println(service.isShutdown());//判断线程池是否被关闭了
        System.out.println(service);
        
        TimeUnit.SECONDS.sleep(5);
        System.out.println(service.isTerminated());
        System.out.println(service.isShutdown());
        System.out.println(service);
    }
}

运行结果：

　　2.六种线程池

　　　　1）FixedThreadPool　　固定线程数量的线程池

　　　　2）cachedThreadPool　　假定刚开始一个线程都没有，来一个任务开启一个线程，如果再来一个任务，线程池中有空闲线程就将任务分配给空闲线程，如此往复直到起到电脑能支撑限度为止，默认一个线程空闲超过一分钟就自动销毁

　　　　3）singleThreadPool　　线程池中只有一个线程

　　　　4）scheduledThreadPool　　定时器线程池

　　　　5）workStealingPool　　工作窃取线程池，当前线程池有线程空闲，会自动去寻找任务执行，默认根据CPU核数启动默认线程数目线程，是精灵（demon）线程（守护线程、后台线程），主函数不阻塞的话看不到输出，本质上是ForkJoinPool实现的

　　　　6）ForkJoinPool　　分叉合并线程，大任务可以切分成许多的小任务，如果小任务还是太大的话还可以继续分，分的可以了就将小任务合并，最后产生一个总的结果（有点像归并排序）

　　　　　  ForkJoinTask从RecursiveAction和RecursiveTask继承

　　　　　　a.RecursiveAction　　无返回值

/**
 * 对数组中所有数求和
 * @author zhangqi
 *
 */
public class T12_ForkJoinPool {
    static int[] nums = new int[1000000];
    static final int MAX_NUM = 50000;
    static Random r = new Random();
    
    static {
        for(int i=0; i<nums.length; i++) {
            nums[i] = r.nextInt(100);
        }
        
        System.out.println(Arrays.stream(nums).sum()); //stream api 
    }
    
    
    static class AddTask extends RecursiveAction { 
        
        int start, end;
        
        AddTask(int s, int e) {
            start = s;
            end = e;
        }

        @Override
        protected void compute() {
            
            if(end-start <= MAX_NUM) {
                long sum = 0L;
                for(int i=start; i<end; i++) sum += nums[i];
                System.out.println("from:" + start + " to:" + end + " = " + sum);
            } else {
            
                int middle = start + (end-start)/2;
                
                AddTask subTask1 = new AddTask(start, middle);
                AddTask subTask2 = new AddTask(middle, end);
                subTask1.fork();
                subTask2.fork();
            }
            
            
        }
        
    }
    
    
    /*static class AddTask extends RecursiveTask<Long> { 
        
        int start, end;
        
        AddTask(int s, int e) {
            start = s;
            end = e;
        }

        @Override
        protected Long compute() {
            
            if(end-start <= MAX_NUM) {
                long sum = 0L;
                for(int i=start; i<end; i++) sum += nums[i];
                return sum;
            } 
            
            int middle = start + (end-start)/2;
            
            AddTask subTask1 = new AddTask(start, middle);
            AddTask subTask2 = new AddTask(middle, end);
            subTask1.fork();
            subTask2.fork();
            
            return subTask1.join() + subTask2.join();
        }
        
    }*/
    
    public static void main(String[] args) throws IOException {
        ForkJoinPool fjp = new ForkJoinPool();
        AddTask task = new AddTask(0, nums.length);
        fjp.execute(task);
        //long result = task.join();
        //System.out.println(result);
        
        System.in.read();
        
    }
}

运行结果：

　　　　　　b.RescursiveTask　　有返回值的递归任务

/**
 * 对数组中所有数求和
 * @author zhangqi
 *
 */
public class T12_ForkJoinPool {
    static int[] nums = new int[1000000];
    static final int MAX_NUM = 50000;
    static Random r = new Random();
    
    static {
        for(int i=0; i<nums.length; i++) {
            nums[i] = r.nextInt(100);
        }
        
        System.out.println(Arrays.stream(nums).sum()); //stream api 
    }
    
    
/*    static class AddTask extends RecursiveAction { 
        
        int start, end;
        
        AddTask(int s, int e) {
            start = s;
            end = e;
        }

        @Override
        protected void compute() {
            
            if(end-start <= MAX_NUM) {
                long sum = 0L;
                for(int i=start; i<end; i++) sum += nums[i];
                System.out.println("from:" + start + " to:" + end + " = " + sum);
            } else {
            
                int middle = start + (end-start)/2;
                
                AddTask subTask1 = new AddTask(start, middle);
                AddTask subTask2 = new AddTask(middle, end);
                subTask1.fork();
                subTask2.fork();
            }
            
            
        }
        
    }*/
    
    
    static class AddTask extends RecursiveTask<Long> { 
        
        int start, end;
        
        AddTask(int s, int e) {
            start = s;
            end = e;
        }

        @Override
        protected Long compute() {
            
            if(end-start <= MAX_NUM) {
                long sum = 0L;
                for(int i=start; i<end; i++) sum += nums[i];
                return sum;
            } 
            
            int middle = start + (end-start)/2;
            
            AddTask subTask1 = new AddTask(start, middle);
            AddTask subTask2 = new AddTask(middle, end);
            subTask1.fork();
            subTask2.fork();
            
            return subTask1.join() + subTask2.join();
        }
        
    }
    
    public static void main(String[] args) throws IOException {
        ForkJoinPool fjp = new ForkJoinPool();
        AddTask task = new AddTask(0, nums.length);
        fjp.execute(task);
        long result = task.join();
        System.out.println(result);
        
        //System.in.read();
        
    }
}

运行结果：

  补充:

　　ThreadPoolExecutor线程池执行器（自定义线程池，许多线程池的底层实现，ForkJoinPool不是由它实现）