Java并发工具类CountDownLatch CyclicBarrier Semaphore，Exchanger，fork/join

r囧r小猫 2022-05-17 10:08 181阅读 0赞

**目录**

1.CountDownLatch

1.1 使用场景

1.2 案例

1.3 原理

1.3.1 await方法

1.3.2 countdown方法

2.CyclicBarrier

2.1 CyclicBarrier和CountDownLatch的区别

3.Semaphore

3.1 使用场景

3.2 案例

3.3 原理

4.Exchanger

5.fork/join

--------------------

JDK并发包中提供几个类，CountDownLatch、CyclicBarrier、Semaphore工具类提供一种并发流程控制的手段，而Exchanger工具类则提供了线程间交换数据。

# **1.CountDownLatch** #

相当于线程中的方法join方法，不过功能更多。CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后，计数器的值就会减一。当计数器的值为0时，表示所有的线程都已经完成了任务，然后在CountDownLatch上等待的线程就可以恢复执行任务。

### **1.1 使用场景** ###

CountDownLatch典型用法1：某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为n `new CountDownLatch(n)` ，每当一个任务线程执行完毕，就将计数器减1 `countdownlatch.countDown()`，当计数器的值变为0时，在CountDownLatch上 `await()` 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。

CountDownLatch典型用法2：实现多个线程开始执行任务的最大**并行性**。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计数器初始化为1，多个线程在开始执行任务前首先 `coundownlatch.await()`，当主线程调用 `countDown()` 时，计数器变为0，多个线程同时被唤醒

源代码：

//构造器
    public CountDownLatch(int count) {  };  //参数count为计数值
    
    //重要的方法
    public void await() throws InterruptedException { };   //调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };  //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
    public void countDown() { };  //将count值减1

### **1.2 案例** ###

public class Test {
         public static void main(String[] args) {   
             final CountDownLatch latch = new CountDownLatch(2);
     
             new Thread(){
                 public void run() {
                     try {
                         System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
                        Thread.sleep(3000);
                        System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
                        latch.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                 };
             }.start();
     
             new Thread(){
                 public void run() {
                     try {
                         System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
                         Thread.sleep(3000);
                         System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
                         latch.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                 };
             }.start();
     
             try {
                 System.out.println("等待2个子线程执行完毕...");
                latch.await();
                System.out.println("2个子线程已经执行完毕");
                System.out.println("继续执行主线程");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
         }
    }
    
    执行结果为：
    
    线程Thread-0正在执行
    线程Thread-1正在执行
    等待2个子线程执行完毕...
    线程Thread-0执行完毕
    线程Thread-1执行完毕
    2个子线程已经执行完毕
    继续执行主线程

### **1.3 原理** ###

CountDownLatch类存在一个内部类Sync，它是一个同步工具，继承了AbstractQueuedSynchronizer。很显然，CountDownLatch实际上是使得线程阻塞了，既然涉及到阻塞，就一定涉及到AQS队列。

### **1.3.1 await方法** ###

**   ** await方法会使得当前线程在countdownlatch倒计时到0之前一直等待，除非线程别中断；从源码中可以得知await  
方法会转发到Sync的acquireSharedInterruptibly

public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); }

**    acquireSharedInterruptibly**

这块代码主要是判断当前线程是否获取到了共享锁; AQS有两种锁类型，一种是共享锁，一种是独占锁，在这里用的是共享锁； 为什么要用共享锁，因为CountDownLatch可以多个线程同时通过。

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
      if (Thread.interrupted()) //判断线程是否中断
            throw new InterruptedException();
      if (tryAcquireShared(arg) < 0) //如果等于0则返回1，否则返回-1，返回-1表示需要阻塞
            doAcquireSharedInterruptibly(arg);  //阻塞的时候调用这个方法
    }
    //在这里，state的意义是count，如果计数器为0，表示不需要阻塞，否则，只有在满足条件的情况下才会被唤醒
    
    
    
    //判断当前state值是否为0
    protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
    }

**    doAcquireSharedInterruptibly**

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
      final Node node = addWaiter(Node.SHARED); //创建一个共享模式的节点添加到队列中
      boolean failed = true;
      try {
        for (;;) { //自旋等待共享锁释放，也就是等待计数器等于0。
           final Node p = node.predecessor(); //获得当前节点的前一个节点
           if (p == head) {
              int r = tryAcquireShared(arg);//就判断尝试获取锁
              if (r >= 0) {//r>=0表示计数器已经归零了，则释放当前的共享锁
                setHeadAndPropagate(node, r);
                p.next = null; // help GC
                failed = false;
                return;
             }
          }
    
          //当前节点不是头节点，则尝试让当前线程阻塞，第一个方法是判断是否需要阻塞，第二个方法是阻塞
          if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
             throw new InterruptedException();
       }
     } finally {
        if (failed)
          cancelAcquire(node);
     }
    }

**setHeadAndPropagate**

private void setHeadAndPropagate(Node node, int propagate) {
      Node h = head; // 记录头节点
      setHead(node); //设置当前节点为头节点
    
      //前面传过来的propagate是1，所以会进入下面的代码
      if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
    
          Node s = node.next; //获得当前节点的下一个节点，如果下一个节点是空表示当前节点为最后一个节
    点，或者下一个节点是share节点
    
         if (s == null || s.isShared())
            doReleaseShared(); //唤醒下一个共享节点
     }
    }

![20190119164844237.png][]

**      doReleaseShared 释放共享锁，通知后面的节点**

private void doReleaseShared() {
       for (;;) {
        Node h = head; //获得头节点
        if (h != null && h != tail) { //如果头节点不为空且不等于tail节点
          int ws = h.waitStatus;
          if (ws == Node.SIGNAL) { //头节点状态为SIGNAL，
            if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) //修改当前头节点的状态为0,避免下次再进入到这个里面
              continue;       // loop to recheck cases
            unparkSuccessor(h); //释放后续节点
         }
    
          else if (ws == 0 &&
              !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
            continue;         // loop on failed CAS
       }
        if (h == head)          // loop if head changed
          break;
     }
    }

### **1.3.2 countdown方法** ###

**        **以共享模式释放锁，并且会调用tryReleaseShared函数，根据判断条件也可能会调用doReleaseShared函数

**  releaseShared**

public final boolean releaseShared(int arg) {
      if (tryReleaseShared(arg)) { //如果为true，表示计数器已归0了
        doReleaseShared(); //唤醒处于阻塞的线程
        return true;
     }
      return false;
    }

**    tryReleaseShared**

这里主要是对state做原子递减，其实就是我们构造的CountDownLatch的计数器，如果等于0返回true，否则返回  
false

protected boolean tryReleaseShared(int releases) {
      // Decrement count; signal when transition to zero
      for (;;) {
        int c = getState();
        if (c == 0)
          return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
          return nextc == 0;
     }
    }

# **2.CyclicBarrier** #

**          **CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

源代码：

//支持两个构造器，parties表示屏障拦截的线程数量，barrierAction表示在线程到达屏障时，优先执行barrierAction线程
    public CyclicBarrier(int parties, Runnable barrierAction) {
    }
     
    public CyclicBarrier(int parties) {
    }
    
    //用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；
    public int await() throws InterruptedException, BrokenBarrierException { };
    
    //让这些线程等待一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。
    public int await(long timeout, TimeUnit unit)throws
    InterruptedException,BrokenBarrierException,TimeoutException { };

例子：

public class CyclicBarrierTest {
    
        static CyclicBarrier c = new CyclicBarrier(2);
    
        public static void main(String[] args) {
    
            new Thread(new Runnable() {
                @Override
                public void run() {
    
                    try {
                        c.await();
                    } catch (Exception e) {
    
                    }
                    System.out.println(1);
                }
            }).start();
    
            try {
                c.await();
            } catch (Exception e) {
            }
            System.out.println(2);
        }
    }
    
    输出：
    1
    2
    或者输出：
    2
    1

如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)则主线程和子线程会永远等待，因为没有第三个线程执行await方法，即没有第三个线程到达屏障，所以之前到达屏障的两个线程都不会继续执行。

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction)，**用于在线程到达屏障时，优先执行barrierAction**，方便处理更复杂的业务场景。代码如下：

public class CyclicBarrierTest2 {
    	static CyclicBarrier c = new CyclicBarrier(2, new A());
    	public static void main(String[] args) {
    		new Thread(new Runnable() {
    			@Override
    			public void run() {
    				try {
    					c.await();
    				} catch (Exception e) {
    
    				}
    				System.out.println(1);
    			}
    		}).start();
    		try {
    			c.await();
    		} catch (Exception e) {
    
    		}
    		System.out.println(2);
    	}
    
    	static class A implements Runnable {
    		@Override
    		public void run() {
    			System.out.println(3);
    		}
    	}
    }
    
    输出结果是：
    3
    1
    2

### 2.1 CyclicBarrier和CountDownLatch的区别 ###

*  CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景，比如如果计算发生错误，可以重置计数器，并让线程们重新执行一次。
 *  CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true。

# **3.Semaphore** #

**         ** 控制并发线程数的semaphore,　Semaphore翻译成字面意思为 信号量，Semaphore可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。如下所示代码中虽然有30个线程在执行，但是只允许10个并发执行，Semaphore(int permits) 接受一个整形数字，表示可用的许可证数量，Semaphore(10)表示允许10个线程获取许可证，也就是并发数为10.

### 3.1 使用场景 ###

可以实现对某些接口访问的限流

### 3.2 案例 ###

public class Test {
        private static final int THREAD_COUTN = 30;
        private static ExecutorService threadPool= Executors.newFixedThreadPool(30);
        private static Semaphore s = new Semaphore(10);
        public static void main(String[] args) {
            for (int i =0;i<THREAD_COUTN;i++){
                
                threadPool.execute(() ->{
                    try {
                        s.acquire();
                        System.out.println(Thread.currentThread().getName()+" test");
                        s.release();
                    }catch (InterruptedException e){
                    }
                });
            }
            threadPool.shutdown();
        }
    }

### 3.3 原理 ###

semaphore也是基于AQS来实现的，内部使用state表示许可数量；它的实现方式和CountDownLatch的差异点在  
于acquireSharedInterruptibly中的tryAcquireShared方法的实现，这个方法是在Semaphore方法中重写的

**acquireSharedInterruptibly**

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
      if (Thread.interrupted())
        throw new InterruptedException();
      if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
    
    }

**tryAcquireShared**

在semaphore中存在公平和非公平的方式，和重入锁是一样的，如果通过FairSync表示公平的信号量、NonFairSync表示非公平的信号量；公平和非公平取决于是否按照FIFO队列中的顺序去分配Semaphore所维护的许可，我们来看非公平锁的实现

**nonfairTryAcquireShared**

自旋去获得一个许可，如果许可获取失败，也就是remaining<0的情况下，让当前线程阻塞

final int nonfairTryAcquireShared(int acquires) {
     for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
          compareAndSetState(available, remaining))
          return remaining;
     }
    }

**releaseShared**

releaseShared方法的逻辑也很简单，就是通过线程安全的方式去增加一个许可，如果增加成功，则触发释放一个共享锁，也就是让之前处于阻塞的线程重新运行。

public final boolean releaseShared(int arg) {
      if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
     }
      return false;
    }

**增加令牌数**

protected final boolean tryReleaseShared(int releases) {
      for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // overflow
          throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
          return true;
     }
    }

# **4.Exchanger** #

1.实现原理

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。因此使用Exchanger的重点是成对的线程使用exchange()方法，当有一对线程达到了同步点，就会进行交换数据。因此该工具类的线程对象是成对的。

Exchanger类提供了两个方法，String exchange(V x):用于交换，启动交换并等待另一个线程调用exchange；String exchange(V x,long timeout,TimeUnit unit)：用于交换，启动交换并等待另一个线程调用exchange，并且设置最大等待时间，当等待时间超过timeout便停止等待。

# 5.fork/join #

采用单任务多线程递归计算结果，一分二的无线递归，知道满足最小临界点；然后拼凑左右两个节点的计算结果。

这里的线程数默认（ForkJoinPool）是当前CPU的数量。直接看实战：

public class ParseResultTask extends RecursiveTask<Boolean> {
        
        /**临界值：用来拆分任务数量 */
        private static final int THREASHOLD = 100;
        //起始点
        private int start;
        //结束点
        private int end;
        /**需要处理的数据*/
        List<String> list;
    
        public ParseResultTask(int start, int end ,List<String> list) {
            this.start = start;
            this.end = end;
            this.list = list;
        }
    
        @Override
        protected Boolean compute() {
            //当数量小于临界值，就做相应的数据处理
            if (end - start < THREASHOLD){
                List<String> arrayList = this.list.subList(start, end);
                System.out.println("分段"+ arrayList);
                return false; //这里简单的返回了false,自定义相应的逻辑
            }else {
                //这里大于临界值，继续做拆分成两个子任务处理，然后把两个任务的处理结果做合并
                int middle = (start + end )/2;
                ParseResultTask left = new ParseResultTask(start, middle, list);
                ParseResultTask right = new ParseResultTask(middle, end, list);
                left.fork();
                right.fork();
                Boolean boolLeft = left.join();
                Boolean boolRight = right.join();
                return boolLeft && boolRight;
            }
        }
    }

public class TestFork  {
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
    
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            List<String> list = new ArrayList<>();
            for (int i = 0; i < 900; i++) {
                list.add(""+i);
            }
    
            ForkJoinTask<Boolean> submit = forkJoinPool.submit(new ParseResultTask(0, list.size(), list));
            System.out.println(submit.get());
        }
    }

参考 [https://blog.csdn.net/carson0408/article/details/79477280][https_blog.csdn.net_carson0408_article_details_79477280]

[20190119164844237.png]: /images/20220517/2db50941327c4d0883b1fbd253a71d77.png
[https_blog.csdn.net_carson0408_article_details_79477280]: https://blog.csdn.net/carson0408/article/details/79477280