java并发包之ExecutorCompletionService、Exchanger、Phaser使用

ExecutorCompletionService类

当我们通过Executor提交一组并发执行的任务，并且希望在每一个任务完成后能立即得到结果，有两种方式可以采取：

• 方式一：

通过一个list来保存一组future，然后在循环中轮训这组future,直到每个future都已完成。如果我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况，那么在调用get方式时，需要将超时时间设置为0 

      static class Task implements Callable<String> {
      private int i;
      public Task(int i) {
      this.i = i;
       }
      @Override
      public String call() throws Exception {
       Thread.sleep(1000);
      return Thread.currentThread().getName() + "执行完任务" + i;
       }
       }
  
 

      private static int numThread = 5;
        //手动创建线程池
       ThreadPoolExecutor threadPoolExecutor =
      new ThreadPoolExecutor(5, 15, 200, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(30));
      List<Future<String>> resultList = new ArrayList<>(5);
      for (int i = 1; i <= numThread; i++) {
       Future<String> future = threadPoolExecutor.submit(new Task(i));
       resultList.add(future);
       }
      while (numThread > 0) {
      if (!CollectionUtils.isEmpty(resultList)) {
      for (Future<String> future : resultList) {
      String result = null;
      try {
       result = future.get(0, TimeUnit.SECONDS);
       } catch (InterruptedException e) {
       e.printStackTrace();
       } catch (ExecutionException e) {
       e.printStackTrace();
       } catch (TimeoutException e) {
      // 超时异常直接忽略
       }
      if (!StringUtils.isEmpty(result)) {
       resultList.remove(future);
       numThread--;
       System.out.println(result);
      // 此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决）
      break;
       }
       }
       }
       }
  
 

• 方式二：

第一种方式显得比较繁琐，通过使用ExecutorCompletionService，则可以达到代码最简化的效果。

       //手动创建线程池
      		ThreadPoolExecutor threadPoolExecutor =
      new ThreadPoolExecutor(5, 15, 200, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(30));
       ExecutorCompletionService<String> executorCompletionService = new ExecutorCompletionService<>(threadPoolExecutor);
      for (int i = 1; i <= numThread; i++) {
       executorCompletionService.submit(new Task(i));
      		}
      for(int i=1;i<=numThread;i++){
     			try {
       System.out.println(executorCompletionService.take().get());
      			} catch (InterruptedException e) {
       e.printStackTrace();
      			} catch (ExecutionException e) {
       e.printStackTrace();
      			}
      		}
       threadPoolExecutor.shutdown();
  
 

 运行结果：

Exchanger类

Exchanger 是 JDK 1.5 开始提供的一个用于两个工作线程之间交换数据的封装工具类，简单说就是一个线程在完成一定的事务后想与另一个线程交换数据，则第一个先拿出数据的线程会一直等待第二个线程，直到第二个线程拿着数据到来时才能彼此交换对应数据。其定义为 Exchanger<V> 泛型类型，其中 V 表示可交换的数据类型，对外提供的接口很简单，具体如下：

• Exchanger()：无参构造方法。

• V exchange(V v)：等待另一个线程到达此交换点（除非当前线程被中断），然后将给定的对象传送给该线程，并接收该线程的对象。

• V exchange(V v, long timeout, TimeUnit unit)：等待另一个线程到达此交换点（除非当前线程被中断或超出了指定的等待时间），然后将给定的对象传送给该线程，并接收该线程的对象。

可以看出，当一个线程到达 exchange 调用点时，如果其他线程此前已经调用了此方法，则其他线程会被调度唤醒并与之进行对象交换，然后各自返回；如果其他线程还没到达交换点，则当前线程会被挂起，直至其他线程到达才会完成交换并正常返回，或者当前线程被中断或超时返回。

      static class Producer extends Thread {
     		private Exchanger<Integer> exchanger;
     		private static int data = 0;
     		Producer(String name, Exchanger<Integer> exchanger) {
     			super("Producer-" + name);
     			this.exchanger = exchanger;
      		}
     		@Override
      		public void run() {
     			for (int i=1; i<5; i++) {
      try {
      TimeUnit.SECONDS.sleep(1);
       data = i;
      System.out.println(getName()+" 交换前:" + data);
       data = exchanger.exchange(data);
      System.out.println(getName()+" 交换后:" + data);
       } catch (InterruptedException e) {
       e.printStackTrace();
       }
      			}
      		}
      	}
      	static class Consumer extends Thread {
     		private Exchanger<Integer> exchanger;
     		private static int data = 0;
     		Consumer(String name, Exchanger<Integer> exchanger) {
     			super("Consumer-" + name);
     			this.exchanger = exchanger;
      		}
     		@Override
      		public void run() {
     			while (true) {
       data = 0;
      System.out.println(getName()+" 交换前:" + data);
      try {
      TimeUnit.SECONDS.sleep(1);
       data = exchanger.exchange(data);
       } catch (InterruptedException e) {
       e.printStackTrace();
       }
      System.out.println(getName()+" 交换后:" + data);
      			}
      		}
      	}
  
 

     	public static void main(String[] args) throws InterruptedException {
      	Exchanger<Integer> exchanger = new Exchanger<Integer>();
     		new Producer("", exchanger).start();
     		new Consumer("", exchanger).start();
      		TimeUnit.SECONDS.sleep(7);
      		System.exit(-1);
       }
  
 

运行结果： 

 Phaser 类

java多线程技术提供了Phaser工具类，Phaser表示“阶段器”，用来解决控制多个线程分阶段共同完成任务的情景问题。其作用相比CountDownLatch和CyclicBarrier更加灵活，例如有这样的一个题目：5个学生一起参加考试，一共有三道题，要求所有学生到齐才能开始考试，全部同学都做完第一题，学生才能继续做第二题，全部学生做完了第二题，才能做第三题，所有学生都做完的第三题，考试才结束。分析这个题目：这是一个多线程（5个学生）分阶段问题（考试考试、第一题做完、第二题做完、第三题做完），所以很适合用Phaser解决这个问题。

      package concurrent.phaser;
      import java.util.concurrent.Phaser;
      /**
       * 比赛阶段器
       */
      public class GamePhaser extends Phaser {
      /**
       * 当一个阶段的所有线程都到达时 , 执行该方法, 此时 phase自动加1
       * @param phase
       * @param registeredParties
       * @return
       */
      @Override
      protected boolean onAdvance(int phase, int registeredParties) {
       switch (phase) {
      case 0 :
      System.out.println("预赛完成");
      return false;
      case 1:
      System.out.println("初赛完成");
      return false;
      case 2:
      System.out.println("决赛完成");
      return false;
      default:
      return true;
       }
       }
      }
  
 

      package concurrent.phaser;
      import java.util.concurrent.Phaser;
      /**
       * 运动员类
       */
      public class Runner implements Runnable {
      private Phaser phaser;
      public Runner(Phaser phaser) {
      this.phaser = phaser;
       }
      @Override
      public void run() {
      /**
       * 参加预赛
       */
       System.out.println("选手-"+Thread.currentThread().getName()+":参加预赛");
      /**
       * 预赛阶段-----执行这个方法的话会等所有的选手都完成了之后再继续下面的方法
       */
       phaser.arriveAndAwaitAdvance();
      /**
       * 参加初赛
       */
       System.out.println("选手-"+Thread.currentThread().getName()+":参加初赛");
      /**
       * 初赛阶段-----执行这个方法的话会等所有的选手都完成了之后再继续下面的方法
       */
       phaser.arriveAndAwaitAdvance();
      /**
       * 参加决赛
       */
       System.out.println("选手-"+Thread.currentThread().getName()+":参加决赛");
      /**
       * 决赛阶段-----执行这个方法的话会等所有的选手都完成了之后再继续下面的方法
       */
       phaser.arriveAndAwaitAdvance();
       }
      }
  
 

      package concurrent.phaser;
      /**
       * 比赛开始
       */
      public class RunnerGame {
      public static void main(String[] args) {
      int runnerNum = 4;
       GamePhaser gamePhaser = new GamePhaser();
      /**
       * 注册一次表示phaser维护的线程个数
       */
       gamePhaser.register();
      for (int i = 0; i < runnerNum;  i++ ) {
      /**
       * 注册一次表示phaser维护的线程个数
       */
       gamePhaser.register();
      new Thread(new Runner(gamePhaser)).start();
       }
      /**
       * 后续阶段主线程就不参加了
       */
       gamePhaser.arriveAndDeregister();
       }
      }
  
 

运行结果：

选手-Thread-0:参加预赛
选手-Thread-1:参加预赛
选手-Thread-3:参加预赛
选手-Thread-2:参加预赛
预赛完成
选手-Thread-2:参加初赛
选手-Thread-0:参加初赛
选手-Thread-3:参加初赛
选手-Thread-1:参加初赛
初赛完成
选手-Thread-1:参加决赛
选手-Thread-2:参加决赛
选手-Thread-0:参加决赛
选手-Thread-3:参加决赛
决赛完成

文章来源: blog.csdn.net，作者：血煞风雨城2018，版权归原作者所有，如需转载，请联系作者。

原文链接：blog.csdn.net/qq_31905135/article/details/108995394