本文共 8202 字，大约阅读时间需要 27 分钟。

等待多线程完成的CountDownLatch

• CountDownLatch允许一个或多个线程等待其他线程完成操作。
• 他的使用场景比如我么需要计算数据, 这个数据很庞大, 所以我们需要使用十几个线程去计算这些数据, 然后在使用一个主线程去汇总这些数据, 那么这个主线程什么时候去启动让他去执行呢? 那就必须等着十几个线程只要只算出来结果就通知一声主线程, 然后继续他们的工作, 当主线程收到所有的线程的通知后就会统计这些数据, 得到安全准确的数据总和
• 演示

public class Demo1 {
       public static void main(String[] args) throws InterruptedException {
           //参数4表示必须受到4次通知才可以        CountDownLatch countDownLatch = new CountDownLatch(4);        MyRunnable myRunnableA = new MyRunnable(1, 4, countDownLatch);        MyRunnable myRunnableB = new MyRunnable(5, 10, countDownLatch);        //创建线程去执行任务        Thread a = new Thread(myRunnableA, "A");        a.start();        Thread b = new Thread(myRunnableB, "B");        b.start();        //在这里进行等待, 直到受到4次通知才不等待        countDownLatch.await();        int result = myRunnableA.getResult() + myRunnableB.getResult();        System.out.println("1~10的和为: " + result);    }}class MyRunnable implements Runnable {
       private int start;    private int end;    private int result;    private CountDownLatch countDownLatch;    public MyRunnable(int start, int end, CountDownLatch countDownLatch) {
           this.start = start;        this.end = end;        this.countDownLatch = countDownLatch;    }    @Override    public void run() {
           //执行一段计算任务        for (int i = start; i <= end; i++) {
               result += i;        }        //通知一次        countDownLatch.countDown();        //一个线程可以通知任意次        countDownLatch.countDown();    }    public int getResult() {
           return result;    }}

• CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成也就是让CoutDownLatch收到N次通知，这里就传入N, 但是这个N一般都是大于0的, 因为在执行CoutDownLatch的await方法内部就是判断计数器是不是等于0, 只要大于0就让当前线程等待, 否则当前线程就不等待直接往下执行
• 当我们调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法 会阻塞当前线程，直到N变成零。由于countDown方法可以用在任何地方，所以这里说的N个 点，可以是N个线程，也可以是1个线程里的N个执行步骤。用在多个线程时，只需要把这个 CountDownLatch的引用传递到线程里即可。
• 如果某个线程处理得比较慢，我们不可能让主线程一直等待，所以可以使 用另外一个带指定时间的await方法——await（long time，TimeUnit unit），这个方法等待特定时 间后，就会不再阻塞当前线程。

同步屏障CyclicBarrier

• 它要做的事情是，让一 组线程到达一个屏障（也可以叫同步点）时直接await后被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行.
• 就像几个特战队员一样, 不能一个人往前直冲, 最前面的那个走一段距离就得停下来等待后面的队员都过来了才可以继续往前走.
• 演示

public class Demo2 {
       public static void main(String[] args) {
           //三表示他这一组有三个线程需要等待, 只有三个线程都在执行了await, 他们才会继续执行        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);        RunnableDemo runnableDemo = new RunnableDemo(cyclicBarrier);        //创建三个线程        Thread a = new Thread(runnableDemo, "A");        Thread b = new Thread(runnableDemo, "B");        Thread c = new Thread(runnableDemo, "C");        a.start();        b.start();        c.start();    }}class RunnableDemo implements Runnable {
       private CyclicBarrier cyclicBarrier;    public RunnableDemo(CyclicBarrier cyclicBarrier) {
           this.cyclicBarrier = cyclicBarrier;    }    @Override    public void run() {
           System.out.println(Thread.currentThread().getName() + " 进入run方法");        try {
               cyclicBarrier.await();            System.out.println(Thread.currentThread().getName() + " 离开run方法");        } catch (InterruptedException | BrokenBarrierException e) {
               e.printStackTrace();        }    }}

• CyclicBarrier默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线程数 量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞, 直到阻塞的线程达到parties, 他们才会继续执行。
• 并且我们从执行结果可以知道, 当这三个线程同时开始执行的时候是不公平的, 也就是它不会保证先等待的先执行, 而是由CPU去任意调度
• CyclicBarrier还提供一个更高级的构造函数CyclicBarrier（int parties，Runnable barrier- Action），用于在线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景

public static void main(String[] args) {
           //三表示他这一组有三个线程需要等待, 只有三个线程都在执行了await, 他们才会继续执行        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() {
               @Override            public void run() {
                   System.out.println("AAA");            }        });        RunnableDemo runnableDemo = new RunnableDemo(cyclicBarrier);        //创建三个线程        Thread a = new Thread(runnableDemo, "A");        Thread b = new Thread(runnableDemo, "B");        Thread c = new Thread(runnableDemo, "C");        a.start();        b.start();        c.start();    }

CyclicBarrier和CountDownLatch的区别

1. CountDownLatch的countDown的使用时任意的, 你可以在一个线程, 也可以多个, 而且在一个线程中使用的次数也是随意的, 但在CyclicBarrier中, 一个线程只能使用一个await
2. CountDownLatch在线程中使用了countDown后不会使线程进入阻塞, 而是继续执行, 但在CyclicBarrier中执行了await线程就会加一道屏障不让其继续执行
3. CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重 置

控制并发线程数的Semaphore

• Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以 保证合理的使用公共资源。
• 把它比作是控制流量的红绿灯。比如××马路要限制流量，只允许同时有一百辆车在这条路上行使，其他的都必须 在路口等待，所以前一百辆车会看到绿灯，可以开进这条马路，后面的车会看到红灯，不能驶 入××马路，但是如果前一百辆中有5辆车已经离开了××马路，那么后面就允许有5辆车驶入马 路，这个例子里说的车就是线程，驶入马路就表示线程在执行，离开马路就表示线程执行完 成，看见红灯就表示线程被阻塞，不能执行。
• Semaphore可以用于做流量控制，特别是公用资源有限的应用场景，比如数据库连接。假 如有一个需求，要读取几万个文件的数据，因为都是IO密集型任务，我们可以启动几十个线程 并发地读取，但是如果读到内存后，还需要存储到数据库中，而数据库的连接数只有10个，这 时我们必须控制只有10个线程同时获取数据库连接保存数据，否则会报错无法获取数据库连 接。这个时候，就可以使用Semaphore来做流量控制
• 演示

public class Demo3 {
       public static void main(String[] args) throws InterruptedException {
           //初始化semaphore        Service service = new Service(10);        //创建线程池让线程池去消耗semaphore资源        ExecutorService pool = Executors.newFixedThreadPool(Service.MAx_THREAD);        for (int i = 0; i < Service.MAx_THREAD; i++) {
               pool.execute(new Runnable() {
                   @Override                public void run() {
                       try {
                           service.getSemaphore().acquire();                        System.out.println("use data");                        System.out.println(new Date().toString());                        Thread.sleep(5000);                        service.getSemaphore().release();                    } catch (InterruptedException e) {
                           e.printStackTrace();                    }                }            });        }        Thread.sleep(1000);        pool.shutdown();    }}class Service {
       public static final int MAx_THREAD = 30;    private Semaphore semaphore;    public Service(int num) {
           this.semaphore = new Semaphore(num);    }    public Semaphore getSemaphore() {
           return semaphore;    }}

• 上图我们发现虽然有30个线程在执行，但是只允许10个并发执行。Semaphore的构造方法 Semaphore（int permits）接受一个整型的数字，表示可用的许可证数量。Semaphore（10）表示允 许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用 Semaphore的acquire()方法获取一个许可证，使用完之后调用release()方法归还许可证。还可以 用tryAcquire()方法尝试获取许可证。

线程间交换数据的Exchanger

• Exchanger用于进行线程间的数据交 换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过 exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也 执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产 出来的数据传递给对方
• 如果两个线程有一个没有执行exchange()方法，则会一直等待，如果担心有特殊情况发 生，避免一直等待，可以使用exchange（V x，longtimeout，TimeUnit unit）设置最大等待时长。
• 演示

public class Demo4 {
       public static void main(String[] args) {
           Exchanger
   
     exchanger = new Exchanger<>();        ThreadA threadA = new ThreadA(exchanger);        ThreadB threadB = new ThreadB(exchanger);        threadA.start();        threadB.start();    }}class ThreadA extends Thread {
       private Exchanger
    
      exchanger;    public ThreadA(Exchanger
     
       exchanger) {
           this.setName("A");        this.exchanger = exchanger;    }    @Override    public void run() {
           String a = "我是Java代码";        try {
               String b = exchanger.exchange(a);            if (a.equals(b)) {
                   System.out.println(Thread.currentThread().getName() + "相同");            } else {
                   System.out.println(Thread.currentThread().getName() + "不相同");            }        } catch (InterruptedException e) {
               e.printStackTrace();        }    }}class ThreadB extends Thread {
       private Exchanger
      
        exchanger;    public ThreadB(Exchanger
       
         exchanger) {
           this.setName("B");        this.exchanger = exchanger;    }    @Override    public void run() {
           String b = "我是Java代码";        try {
               String a = exchanger.exchange(b);            if (b.equals(a)) {
                   System.out.println(Thread.currentThread().getName() + "相同");            } else {
                   System.out.println(Thread.currentThread().getName() + "不相同");            }        } catch (InterruptedException e) {
               e.printStackTrace();        }    }}
       
      
     
    
   

转载地址：http://tjsci.baihongyu.com/