BiBi - 并发编程 -13- 并发工具类

From：Java并发编程的艺术

1. CountDownLatch【等待其它线程完成】

应用场景：主线程等待其他线程执行完后再继续执行。

使用join实现
thread1.start();
thread2.start();
thread1.join(); //或者thread1.join(5000)
thread2.join();
println("执行完成");

join方法的原理：不停地检查join线程是否存活，如果存活则继续等待，直到join线程终止后，调用notifyAll()方法。

while (isAlive()) {
  wait(0); //表示永远等待下去
}
CountDownLatch实现
package com.ljg;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest {
  static CountDownLatch c = new CountDownLatch(2);

  public static void main(String[] args) {
    new Thread(new ThreadTest(c)).start();
    new Thread(new ThreadTest(c)).start();
    try {
      c.await(); //或者c.await(5000)
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("执行完成");
  }

  public static class ThreadTest implements Runnable {
    CountDownLatch c;

    public ThreadTest(CountDownLatch c) {
      this.c = c;
    }

    @Override
    public void run() {
      //处理逻辑
      c.countDown();
    }
  }
}

CountDownLatch要比join方法使用起来更加灵活。【体会】

注意：CountDownLatch不能重新初始化或修改其内部的计数器，即new CountDownLatch(2)中的2，不能被修改。并且这里的计数2，可以是2个线程，也可以是一个线程里的2个执行步骤。

2. CyclicBarrier【同步屏障】

应用场景：多线程计算数据，最后合并计算结果
字面含义：循环使用的屏障。它能让一个线程到达一个屏障【也可以叫同步点】时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续执行。

    //在所有线程到达屏障后，执行barrierAction任务。
    public CyclicBarrier(int parties, Runnable barrierAction) { }
例子：统计Excel中每个sheet表中的用户流水，最后再用barrierAction处理计算结果。
package com.ljg;

import java.util.Map;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class BankWaterService implements Runnable {

  //创建4个屏障，并点定义所有屏障到达后的处理任务
  private CyclicBarrier c = new CyclicBarrier(4, this);
  //启动4个线程
  private Executor executor = Executors.newFixedThreadPool(4);
  private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();

  private void count() {
    for (int i = 0; i < 4; ++i) {
      executor.execute(new Runnable() {
        @Override
        public void run() {
          //记录每个线程的计算结果
          sheetBankWaterCount.put(Thread.currentThread().getName(), 177);
          try {
            //插入一个屏障
            c.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          } catch (BrokenBarrierException e) {
            e.printStackTrace();
          }
        }
      });
    }
  }

  @Override
  public void run() {
    int result = 0;
    for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
      result += sheet.getValue();
    }
    System.out.println("计算结果为：" + result);
  }

  public static void main(String[] args) {
    BankWaterService bankWaterService = new BankWaterService();
    bankWaterService.count();
  }
}
CyclicBarrier与CountDownLatch的区别

CountDownLatch的计数器一旦初始化就不能再改动，而CyclicBarrier的计数器可以通过reset()方法重置。如：当计算结果发生错误，可以重置计数器，并让线程重新执行一次。因此CyclicBarrier能够处理更加复杂的业务场景。

CyclicBarrier可以通过isBroken()判断线程是否被中断。

3. Semaphore【控制并发线程数】

应用场景：用来做流量控制，特别是公共资源有限制的场景，如：数据库连接。跟自定义同步组件TwinsLock功能类似。

例子：允许只有10个并发线程执行。
package com.ljg;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreTest {
  private static final int THREAD_COUNT = 50;
  private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
  private static Semaphore s = new Semaphore(10);

  public static void main(String[] args) {
    for (int i = 0; i < THREAD_COUNT; ++i) {
      threadPool.execute(new Runnable() {
        @Override
        public void run() {
          try {
            s.acquire();
            /*
             *操作数据
             */
            s.release();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      });
    }
    threadPool.shutdown();
  }
}

上述代码中启动了50个线程，但只允许10个线程并发执行，其它线程处于等待状态。

4. Exchanger【线程间交换数据】

一个线程先执行exchange()方法，他会一直等待第二个线程也执行exchange()方法，然后就可以交换数据。

package com.ljg;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExchangerTest {
  private static final Exchanger<String> exgr = new Exchanger<>();
  private static ExecutorService threadPool = Executors.newFixedThreadPool(2);

  public static void main(String[] args) {
    threadPool.execute(new Runnable() {
      @Override
      public void run() {
        String A = "银行流水A";
        try {
          exgr.exchange(A); //也可以设置最大等待时长
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
    threadPool.execute(new Runnable() {
      @Override
      public void run() {
        String B = "银行流水B";
        try {
          String back = exgr.exchange(B);
          System.out.println("从其他线程获取的数据为：" + back);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
  }
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。