Java并发编程—CountDownLatch、Semaphore和CyclicBarrie

冷不防 2024-04-06 12:43 9阅读 0赞

## CountDownLatch、Semaphore和CyclicBarrier概念说明及区别 ##

### 概念 ###

概念说明：

1.  **CountDownLatch**：一般用作多线程倒计时计数器，强制它们等待其他一组（CountDownLatch的初始化决定）任务执行完成。
    
    > 比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。  
    > 有一点要说明的是CountDownLatch初始化后计数器值递减到0的时候，不能再复原的，这一点区别于Semaphore，Semaphore是可以通过release操作恢复信号量的
2.  **CyclicBarrier**：一组线程使用await()指定barrier，所有线程都到达各自的barrier后，再同时执行各自barrier下面的代码
3.  **Semaphore**： 每次semaphore.acquire()，获取一个资源，每次semaphore.acquire(n)，获取n个资源，`当达到semaphore 指定资源数量时就不能再访问线程处于阻塞，必须等其它线程释放资源`

### 区别和相同点 ###

区别如下：

*  **CountDownLatch**：
    
    1.  `减计数方式`，计数==0时释放所有等待的线程
    2.  多线程倒计时计数器，`强制它们等待其他一组（CountDownLatch的初始化决定）任务执行完成`
    3.  `当计数到0时，计数无法被重置`
    4.  每次调用countDown()方法计数减一，`调用await()方法只进行阻塞，对计数没任何影响`
 *  **CyclicBarrier**：
    
    1.  `加计数方式`，计数达到构造方法中参数指定的值时释放所有等待的线程
    2.  组线程`使用await()指定barrier，所有线程都到达各自的barrier后，再同时执行各自barrier下面的代码`
    3.  `计数达到指定值时，计数置为0重新开始`
    4.  只有一个await()方法，`调用await()方法计数加1，若加1后的值不等于构造方法的值，则线程阻塞`。
 *  **Semaphore**：
    
    1.  每次semaphore.acquire()，获取一个资源，每次semaphore.acquire(n)，获取n个资源，`当达到semaphore 指定资源数量时就不能再访问线程处于阻塞，必须等其它线程释放资源`
        
        > semaphore.relase()每次资源一个资源，semaphore.relase(n)每次资源n个资源
    2.  用来`控制同时访问特定资源的线程数量`，它通过`协调各个线程，以保证合理的使用公共资源`

共同点：

1.  **CountDownLatch与CyclikBarrie**r两者的共同点是都具有**await()方法**，并且执行此方法**会引起线程的阻塞**，达到某种条件才能继续执行（这种条件也是两者的不同）
2.  Semaphore，acquire方**获取的资源达到最大数量时，线程再次acquire获取资源时，也会使线程处于阻塞状态**。
3.  三者都有一个int类型参数的构造方法
    
     *  **CountDownLatch、CyclikBarrie**r这个值作为计数用，`达到该次数即释放等待的线程`，
     *  **Semaphore 中所有acquire**获取到的资源`达到这个数，会使得其它线程阻塞`。

## CountDownLatch ##

### 使用场景 ###

使用场景：一个程序中有`N个任务在执行`，我们可以创建`值为N的CountDownLatch`，当`每个任务完成后，调用一下countDown()方法进行递减count值`，再在`主线程中使用await()方法`等待任务执行完成，**等待所有的任务执行完成后，主线程才开始继续执行**。

### 常用方法 ###

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

### 使用 ###

public class CountDownLatchExample {
        
      public static void main(String[] args) throws InterruptedException {
        
        CountDownLatch latch = new CountDownLatch(5);
        Service service = new Service(latch);
        Runnable task = () -> service.exec();
    
        for (int i = 0; i < 5; i++) {
        
          Thread thread = new Thread(task);
          thread.start();
        }
    
        System.out.println("main thread await. ");
        latch.await();
        System.out.println("main thread finishes await. ");
      }
    }
    
    public class Service {
        
      private CountDownLatch latch;
    
      public Service(CountDownLatch latch) {
        
        this.latch = latch;
      }
    
      public void exec() {
        
        try {
        
          System.out.println(Thread.currentThread().getName() + " execute task. ");
          sleep(2);
          System.out.println(Thread.currentThread().getName() + " finished task. ");
        } finally {
        
          latch.countDown();
        }
      }
    
      private void sleep(int seconds) {
        
        try {
        
          TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
        
          e.printStackTrace();
        }
      }
    }

首先声明了一个CountDownLatch对象，并且由主线程创建了5个线程，分别执行任务，在每个任务中，当前线程会休眠2秒。在启动线程之后，主线程调用了CountDownLatch.await()方法，此时，主线程将在此处等待创建的5个线程执行完任务之后才继续往下执行。如下是执行结果：

Thread-0 execute task. 
    Thread-1 execute task. 
    Thread-2 execute task. 
    Thread-3 execute task. 
    Thread-4 execute task. 
    main thread await. 
    Thread-0 finished task. 
    Thread-4 finished task. 
    Thread-3 finished task. 
    Thread-1 finished task. 
    Thread-2 finished task. 
    main thread finishes await.

可以看出，主线程先启动了五个线程，然后主线程进入等待状态，当这五个线程都执行完任务之后主线程才结束了等待，代码中使用try…finally…结构可以保证创建的线程发生异常时CountDownLatch.countDown()方法也会执行，也就保证了主线程不会一直处于等待状态

## Semaphore ##

Semaphore翻译成字面意思为 信号量，Semaphore可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可

### 使用场景 ###

Semaphore其实和锁有点类似，它一般用于控制对某组资源的访问权限

### 常用方法 ###

Semaphore类位于java.util.concurrent包下，它提供了**2个构造器**：

//参数permits表示许可数目，即同时可以允许多少线程进行访问
    public Semaphore(int permits) {
                  
        sync = new NonfairSync(permits);
    }
    //这个多了一个参数fair表示是否是公平的，即等待时间越久的越先获取许可
    public Semaphore(int permits, boolean fair) {
            
        sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
    }

Semaphore类中比较重要的几个方法

//获取一个许可
    public void acquire() throws InterruptedException {
          }     
    //获取permits个许可
    public void acquire(int permits) throws InterruptedException {
         }    
    //释放一个许可
    public void release() {
         }       
    //释放permits个许可   
    public void release(int permits) {
         }

acquire()用来获取一个许可，若 **无许可能够获得**，则`会一直等待，直到获得许可`。  
release()用来**释放许可**。注意，**在释放许可之前，必须先获获得许可**。

上面的4个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法

//尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false
    public boolean tryAcquire() {
         };    
    //尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false
    public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
         }; 
    //尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false 
    public boolean tryAcquire(int permits) {
         }; 
    /尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false
    public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException {
         }; /

### 使用 ###

若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。那么我们就可以通过Semaphore来实现：

public class Test {
        
        public static void main(String[] args) {
        
            int N = 8;            //工人数
            Semaphore semaphore = new Semaphore(5); //机器数目
            for(int i=0;i<N;i++)
                new Worker(i,semaphore).start();
        }
         
        static class Worker extends Thread{
        
            private int num;
            private Semaphore semaphore;
            public Worker(int num,Semaphore semaphore){
        
                this.num = num;
                this.semaphore = semaphore;
            }
             
            @Override
            public void run() {
        
                try {
        
                    semaphore.acquire();
                    System.out.println("工人"+this.num+"占用一个机器在生产...");
                    Thread.sleep(2000);
                    System.out.println("工人"+this.num+"释放出机器");
                    semaphore.release();           
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

执行结果

工人0占用一个机器在生产...
    工人1占用一个机器在生产...
    工人2占用一个机器在生产...
    工人4占用一个机器在生产...
    工人5占用一个机器在生产...
    工人0释放出机器
    工人2释放出机器
    工人3占用一个机器在生产...
    工人7占用一个机器在生产...
    工人4释放出机器
    工人5释放出机器
    工人1释放出机器
    工人6占用一个机器在生产...
    工人3释放出机器
    工人7释放出机器
    工人6释放出机器

## CyclicBarrier ##

### 使用场景 ###

CyclicBarrier可以`用在所有子线程之间互相等待多次的情形`

*  在某种需求中，比如一个大型的任务，常常需要分配好多子任务去执行，只有当所有子任务都执行完成时候，才能执行主任务，这时候，就可以选择CyclicBarrier了。
 *  比如团队旅游，一个团队通常分为几组，每组人走的路线可能不同，但都需要到达某一地点等待团队其它成员到达后才能进行下一站。

### 常用方法 ###

CyclicBarrier类位于java.util.concurrent包下，CyclicBarrier提供2个构造器

public CyclicBarrier(int parties, Runnable barrierAction) {
        }
     
    public CyclicBarrier(int parties) {
        }

*  参数parties指**让多少个线程或者任务等待至barrier状态**；
 *  参数barrierAction为当这些线程都**达到barrier状态时会执行的内容**。

await方法，它有2个重载版本：

public int await() throws InterruptedException, BrokenBarrierException {
         };
    public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException {
         };

*  第一个版本比较常用，用来挂起当前线程，**直至所有线程都到达barrier状态再同时执行后续任务**；
 *  第二个版本是让这些线程等待至一定的时间，**如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务**。

### 使用 ###

#### 所有线程等待到某个状态后执行 ####

假若有若干个线程都要进行写数据操作，并且只有所有线程都完成写数据操作之后，这些线程才能继续做后面的事情，此时就可以利用CyclicBarrier了：

public class Test {
        
        public static void main(String[] args) {
        
            int N = 4;
            CyclicBarrier barrier  = new CyclicBarrier(N);
            for(int i=0;i<N;i++)
                new Writer(barrier).start();
        }
        static class Writer extends Thread{
        
            private CyclicBarrier cyclicBarrier;
            public Writer(CyclicBarrier cyclicBarrier) {
        
                this.cyclicBarrier = cyclicBarrier;
            }
     
            @Override
            public void run() {
        
                System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
                try {
        
                    Thread.sleep(5000);      //以睡眠来模拟写入数据操作
                    System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }catch(BrokenBarrierException e){
        
                    e.printStackTrace();
                }
                System.out.println("所有线程写入完毕，继续处理其他任务...");
            }
        }
    }

执行结果

线程Thread-0正在写入数据...
    线程Thread-3正在写入数据...
    线程Thread-2正在写入数据...
    线程Thread-1正在写入数据...
    线程Thread-2写入数据完毕，等待其他线程写入完毕
    线程Thread-0写入数据完毕，等待其他线程写入完毕
    线程Thread-3写入数据完毕，等待其他线程写入完毕
    线程Thread-1写入数据完毕，等待其他线程写入完毕
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...

每个写入线程执行完写数据操作之后，就在等待其他线程写入操作完毕。当所有线程线程写入操作完毕之后，所有线程就继续进行后续的操作了。

#### 线程执行完后想要执行一些操作 ####

如果想在所有线程写入操作完之后，进行额外的其他操作，可以在CyclicBarrier提供Runnable参数中做相应的处理：

public class Test {
        
        public static void main(String[] args) {
        
            int N = 4;
            CyclicBarrier barrier  = new CyclicBarrier(N,new Runnable() {
        
                @Override
                public void run() {
        
                    System.out.println("当前线程"+Thread.currentThread().getName());   
                }
            });
             
            for(int i=0;i<N;i++)
                new Writer(barrier).start();
        }
        static class Writer extends Thread{
        
            private CyclicBarrier cyclicBarrier;
            public Writer(CyclicBarrier cyclicBarrier) {
        
                this.cyclicBarrier = cyclicBarrier;
            }
     
            @Override
            public void run() {
        
                System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
                try {
        
                    Thread.sleep(5000);      //以睡眠来模拟写入数据操作
                    System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }catch(BrokenBarrierException e){
        
                    e.printStackTrace();
                }
                System.out.println("所有线程写入完毕，继续处理其他任务...");
            }
        }
    }

执行结果

线程Thread-0正在写入数据...
    线程Thread-1正在写入数据...
    线程Thread-2正在写入数据...
    线程Thread-3正在写入数据...
    线程Thread-0写入数据完毕，等待其他线程写入完毕
    线程Thread-1写入数据完毕，等待其他线程写入完毕
    线程Thread-2写入数据完毕，等待其他线程写入完毕
    线程Thread-3写入数据完毕，等待其他线程写入完毕
    当前线程Thread-3
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...
    所有线程写入完毕，继续处理其他任务...

上述代码来源1：[https://www.cnblogs.com/dolphin0520/p/3920397.html][https_www.cnblogs.com_dolphin0520_p_3920397.html]

上述代码来源2：[https://www.jianshu.com/p/128476015902][https_www.jianshu.com_p_128476015902]

[https_www.cnblogs.com_dolphin0520_p_3920397.html]: https://www.cnblogs.com/dolphin0520/p/3920397.html
[https_www.jianshu.com_p_128476015902]: https://www.jianshu.com/p/128476015902