你的位置:首页 > Java教程

[Java教程]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不是由它实现)