CountDownLatch、CyclicBarrier、Semaphore 用法总结

时间  2019-12-07
标签 countdownlatch cyclicbarrier semaphore 用法 总结 栏目 Java 繁體版
原文   原文链接

CountDownLatch

  • CountDownLatch 类位于 java.util.concurrent 包下，利用它能够实现相似计数器的功能。好比有一个任务A，它要等待其余4个任务执行完毕以后才能执行，此时就能够利用CountDownLatch来实现这种功能了。
  • CountDownLatch类只提供了一个构造器：
public CountDownLatch(int count) {  };  //参数count为计数值
  • 而后下面这3个方法是CountDownLatch类中最重要的方法：
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
  • 代码实现
package sychronized;

import static net.mindview.util.Print.*;
import java.util.concurrent.*;

class Task implements Runnable{
    private static int count = 0;
    private final int id = count++;
    final CountDownLatch latch ;
    public Task(CountDownLatch latch){
        this.latch = latch;
    }
    
    @Override
    public void run(){
        try {
            print(this+"正在执行");
            TimeUnit.MILLISECONDS.sleep(3000);
            print(this+"执行完毕");
            latch.countDown();
        } catch (InterruptedException e) {
            print(this + " 被中断");
        }
    }

    @Override
    public String toString() {
        return "Task-"+id;
    }
}

public class Test {
    public static void main(String[] args) {
        final CountDownLatch latch = new CountDownLatch(2);
        ExecutorService exec = Executors.newCachedThreadPool();
        
        exec.execute(new Task(latch));
        exec.execute(new Task(latch));
        
        try {
            print("等待2个子线程执行完毕...");
            long start = System.currentTimeMillis();
            latch.await();
            long end = System.currentTimeMillis();
            
            print("2个子线程已经执行完毕 "+(end - start));
            print("继续执行主线程");
        }catch (InterruptedException e){
            print("主线程被中断");
        }
        exec.shutdown();
    }
}

#输出结果：
等待2个子线程执行完毕...
Task-0正在执行
Task-1正在执行
Task-0执行完毕
Task-1执行完毕
2个子线程已经执行完毕 3049
继续执行主线程

CyclicBarrier

  • 字面意思回环栅栏，经过它能够实现让一组线程等待至某个状态以后再所有同时执行。叫作回环是由于当全部等待线程都被释放之后，CyclicBarrier能够被重用。咱们暂且把这个状态就叫作barrier，当调用await()方法以后，线程就处于barrier了。
  • CyclicBarrier类位于java.util.concurrent包下，CyclicBarrier提供2个构造器：
  • 参数parties指让多少个线程或者任务等待至barrier状态
  • 参数barrierAction为当这些线程都达到barrier状态时会执行的内容
public CyclicBarrier(int parties, Runnable barrierAction) {}
public CyclicBarrier(int parties) {}
  • 而后CyclicBarrier中最重要的方法就是 await 方法，它有2个重载版本：
  • 第一个版本比较经常使用，用来挂起当前线程，直至全部线程都到达barrier状态再同时执行后续任务；
  • 第二个版本是让这些线程等待至必定的时间，若是还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。
public int await() throws InterruptedException, BrokenBarrierException { };
public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException { };
  • 代码展现
package sychronized;

import java.util.Random;
import java.util.concurrent.*;
import static net.mindview.util.Print.*;

class WriteTask implements Runnable{
    private static int count = 0;
    private final int id = count++;
    private CyclicBarrier barrier ;
    private static Random random = new Random(47);
    public WriteTask(CyclicBarrier cyclicBarrier) {
        this.barrier = cyclicBarrier;
    }
    
    @Override
    public void run() {
        print(this+"开始写入数据...");
        try {
            TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));      //以睡眠来模拟写入数据操做
            print(this+"写入数据完毕，等待其余线程写入完毕"+" "+System.currentTimeMillis());
            barrier.await();
        } catch (InterruptedException e) {
            print(this + "is interrupted!");
        }catch(BrokenBarrierException e){
           throw new RuntimeException(e);
        }
        print("全部任务写入完毕，继续处理其余任务... "+System.currentTimeMillis());
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()+"-"+id;
    }
}

public class CyclicBarrierTest {
    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier  = new CyclicBarrier(N);
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i = 0; i < N; ++i){
            exec.execute(new WriteTask(barrier));
        }
        exec.shutdown();
    }
}

#输出结果：
WriteTask-3 开始写入数据...
WriteTask-2 开始写入数据...
WriteTask-1 开始写入数据...
WriteTask-0 开始写入数据...
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512048648904
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512048650042
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512048650209
WriteTask-3 写入数据完毕，等待其余线程写入完毕 1512048652606
全部任务写入完毕，继续处理其余任务... 1512048652607
全部任务写入完毕，继续处理其余任务... 1512048652607
全部任务写入完毕，继续处理其余任务... 1512048652607
全部任务写入完毕，继续处理其余任务... 1512048652607

**java

  • 若是说想在全部线程写入操做完以后，进行额外的其余操做能够为CyclicBarrier提供Runnable参数：

**dom

package sychronized;

import java.util.Random;
import java.util.concurrent.*;
import static net.mindview.util.Print.*;

class WriteTask implements Runnable{
    private static int count = 0;
    private final int id = count++;
    private CyclicBarrier barrier ;
    private static Random random = new Random(47);
    public WriteTask(CyclicBarrier cyclicBarrier) {
        this.barrier = cyclicBarrier;
    }
    
    @Override
    public void run() {
        print(this+" 开始写入数据...");
        try {
            TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));      //以睡眠来模拟写入数据操做
            print(this+" 写入数据完毕，等待其余线程写入完毕"+" "+System.currentTimeMillis());
            barrier.await();
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            print(this + "is interrupted!");
        }catch(BrokenBarrierException e){
           throw new RuntimeException(e);
        }
        print("全部任务写入完毕，继续处理其余任务... "+System.currentTimeMillis()+Thread.currentThread());
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()+"-"+id;
    }
}

public class CyclicBarrierTest {
    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier  = new CyclicBarrier(N, new Runnable() {
            @Override
            public void run() {
                print(Thread.currentThread());
            }
        });
     
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i = 0; i < N; ++i){
            exec.execute(new WriteTask(barrier));
        }
        exec.shutdown();
    }
}
#输出结果为：
WriteTask-3 开始写入数据...
WriteTask-1 开始写入数据...
WriteTask-2 开始写入数据...
WriteTask-0 开始写入数据...
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512049061954
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512049063092
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512049063261
WriteTask-3 写入数据完毕，等待其余线程写入完毕 1512049065657
Thread[pool-1-thread-4,5,main]
全部任务写入完毕，继续处理其余任务... 1512049065668Thread[pool-1-thread-2,5,main]
全部任务写入完毕，继续处理其余任务... 1512049065668Thread[pool-1-thread-1,5,main]
全部任务写入完毕，继续处理其余任务... 1512049065668Thread[pool-1-thread-4,5,main]
全部任务写入完毕，继续处理其余任务... 1512049065668Thread[pool-1-thread-3,5,main]

从结果能够看出，当四个线程都到达barrier状态后，会从四个线程中选择一个线程去执行Runnable。ide

  • 另外CyclicBarrier是能够重用的，看下面这个例子：
package sychronized;

import java.util.Random;
import java.util.concurrent.*;
import static net.mindview.util.Print.*;

class WriteTask implements Runnable{
    private static int count = 0;
    private final int id = count++;
    private CyclicBarrier barrier ;
    private static Random random = new Random(47);
    public WriteTask(CyclicBarrier cyclicBarrier) {
        this.barrier = cyclicBarrier;
    }
    
    @Override
    public void run() {
        
        while (!Thread.interrupted()){
            print(this+" 开始写入数据...");
            try {
                TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));      //以睡眠来模拟写入数据操做
                print(this+" 写入数据完毕，等待其余线程写入完毕"+" "+System.currentTimeMillis());
                barrier.await();
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                print(this + "is interrupted!");
            }catch(BrokenBarrierException e){
                throw new RuntimeException(e);
            }
            print("全部任务写入完毕，继续处理其余任务... "+System.currentTimeMillis());
        }
       
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()+"-"+id;
    }
}

class CyclicBarrierManager implements Runnable{
    private CyclicBarrier barrier ;
    private ExecutorService exec;
    public CyclicBarrierManager(CyclicBarrier barrier, ExecutorService exec,int N){
        this.barrier = barrier ;
        this.exec = exec;
        for (int i = 0; i < N-1; ++i){
            exec.execute(new WriteTask(barrier));
        }
    }
    
    @Override
    public void run(){
        while (!Thread.interrupted()){
            try {
                barrier.await();
            }catch (InterruptedException e){
                print(getClass().getSimpleName()+" 被中断了！");
            }catch (BrokenBarrierException e){
                throw new RuntimeException(e);
            }
        }
    }
}

public class CyclicBarrierTest {
    public static void main(String[] args) throws Exception{
        int N = 4;
        CyclicBarrier barrier  = new CyclicBarrier(N);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new CyclicBarrierManager(barrier,exec,N));
        exec.shutdown();
    }
}

#输出结果：
WriteTask-1 开始写入数据...
WriteTask-2 开始写入数据...
WriteTask-0 开始写入数据...
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512051484365
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512051485503
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512051488068
全部任务写入完毕，继续处理其余任务... 1512051488078
全部任务写入完毕，继续处理其余任务... 1512051488078
WriteTask-2 开始写入数据...
全部任务写入完毕，继续处理其余任务... 1512051488078
WriteTask-1 开始写入数据...
WriteTask-0 开始写入数据...
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512051488513
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512051489045
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512051489945
全部任务写入完毕，继续处理其余任务... 1512051489955
WriteTask-0 开始写入数据...
全部任务写入完毕，继续处理其余任务... 1512051489955
全部任务写入完毕，继续处理其余任务... 1512051489955
WriteTask-2 开始写入数据...
WriteTask-1 开始写入数据...
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512051490155
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512051494477
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512051494823
全部任务写入完毕，继续处理其余任务... 1512051494833
全部任务写入完毕，继续处理其余任务... 1512051494833
WriteTask-0 开始写入数据...
全部任务写入完毕，继续处理其余任务... 1512051494833
WriteTask-1 开始写入数据...
WriteTask-2 开始写入数据...
WriteTask-2 写入数据完毕，等待其余线程写入完毕 1512051494961
WriteTask-0 写入数据完毕，等待其余线程写入完毕 1512051496040
WriteTask-1 写入数据完毕，等待其余线程写入完毕 1512051498121
全部任务写入完毕，继续处理其余任务... 1512051498132
全部任务写入完毕，继续处理其余任务... 1512051498132
WriteTask-1 开始写入数据...
全部任务写入完毕，继续处理其余任务... 1512051498132

Semaphore

  • Semaphore翻译成字面意思为 信号量，Semaphore 能够同时让多个线程同时访问共享资源，经过 acquire() 获取一个许可，若是没有就等待，而 release() 释放一个许可。
  • Semaphore类位于java.util.concurrent包下，它提供了2个构造器：
public Semaphore(int permits) {          //参数permits表示许可数目，即同时能够容许多少线程进行访问
    sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {    //这个多了一个参数fair表示是不是公平的，即等待时间越久的越先获取许可
    sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
}
  • 下面说一下Semaphore类中比较重要的几个方法，首先是acquire()、release()方法：
public void acquire() throws InterruptedException {  }     //获取一个许可
public void acquire(int permits) throws InterruptedException { }    //获取permits个许可
public void release() {}          //释放一个许可
public void release(int permits) {}    //释放permits个许可
  • acquire()用来获取一个许可，若无许可可以得到，则会一直等待，直到得到许可。
  • release()用来释放许可。
  • 注意，在释放许可以前，必须先获得到许可。

　这4个方法都会被阻塞，若是想当即获得执行结果，可使用下面几个方法：ui

public boolean tryAcquire() { };    //尝试获取一个许可，若获取成功，则当即返回true，若获取失败，则当即返回false
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { };  //尝试获取一个许可，若在指定的时间内获取成功，则当即返回true，不然则当即返回false
public boolean tryAcquire(int permits) { }; //尝试获取permits个许可，若获取成功，则当即返回true，若获取失败，则当即返回false
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { }; //尝试获取permits个许可，若在指定的时间内获取成功，则当即返回true，不然则当即返回false
package sychronized;
import java.util.Random;
import java.util.concurrent.*;
import static net.mindview.util.Print.*;

class Worker implements Runnable{
    private static int count = 0;
    private final int id = count++;
    private int finished = 0;
    private Random random = new Random(47);
    private Semaphore semaphore;
    public Worker(Semaphore semaphore){
        this.semaphore = semaphore;
    }
    
    @Override 
    public void run(){
        try {
            while (!Thread.interrupted()){
                semaphore.acquire();
                print(this+" 占用一个机器在生产...   ");
                TimeUnit.MILLISECONDS.sleep(random.nextInt(2000));
                synchronized (this){
                    print(" 已经生产了"+(++finished)+"个产品,"+"释放出机器");
                }
                semaphore.release();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()+"-"+id;
    }
}

public class SemaphoreTest {
    public static void main(String[] args) {
        int N = 8;            //工人数
        Semaphore semaphore = new Semaphore(5); //机器数目
        ExecutorService exec = Executors.newCachedThreadPool();
        for (int i = 0; i < N; ++i){
            exec.execute(new Worker(semaphore));
        }
        exec.shutdown();
    }
}

总结

CountDownLatch和CyclicBarrier都可以实现线程之间的等待，只不过它们侧重点不一样：this

  • CountDownLatch 通常用于某个线程A等待若干个其余线程执行完任务以后，它才执行；
  • CyclicBarrier 通常用于一组线程互相等待至某个状态，而后这一组线程再同时执行；
  • CountDownLatch 是不可以重用的，而 CyclicBarrier 是能够重用的。

Semaphore 其实和锁有点相似，它通常用于控制对 某组 资源的访问权限，而锁是控制对 某个 资源的访问权限。线程

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程