Java并发包之闭锁/栅栏/信号量（转）

时间：2021-03-10 15:16:34
本文转自http://blog.csdn.net/u010942020/article/details/79352560 感谢作者
一、Java多线程总结：
  1. 描述线程的类：Runable和Thread都属于java.lang包。
  2. 内置锁synchronized属于jvm关键字，内置条件队列操作接口Object.wait()/notify()/notifyAll()属于java.lang包。
  3. 提供内存可见性和防止指令重排的volatile属于jvm关键字。
  4. 而java.util.concurrent包(J.U.C)中包含的是java并发编程中有用的一些工具类，包括几个部分： 
    • locks部分：包含在java.util.concurrent.locks包中，提供显式锁(互斥锁和速写锁)相关功能。
    • atomic部分：包含在java.util.concurrent.atomic包中，提供原子变量类相关的功能，是构建非阻塞算法的基础。
    • executor部分：散落在java.util.concurrent包中，提供线程池相关的功能。
    • collections部分：散落在java.util.concurrent包中，提供并发容器相关功能。
    • tools部分：散落在java.util.concurrent包中，提供同步工具类，如信号量、闭锁、栅栏等功能。 
      Java并发包之闭锁/栅栏/信号量（转）
二、同步工具类详解

1、Semaphore信号量：跟锁机制存在一定的相似性，semaphore也是一种锁机制，所不同的是，reentrantLock是只允许一个线程获得锁，而信号量持有多个许可(permits)，允许多个线程获得许可并执行。可以用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。

示例代码：

 5 public class TIJ_semaphore { 6 public static void main(String[] args) { 7 ExecutorService exec = Executors.newCachedThreadPool(); 8 final Semaphore semp = new Semaphore(5); // 5 permits 9 10 for (int index = 0; index < 20; index++) { 11 final int NO = index; 12 Runnable run = new Runnable() { 13 public void run() { 14 try { // if 1 permit avaliable, thread will get a permits and go; if no permit avaliable, thread will block until 1 avaliable 15 semp.acquire(); 16 System.out.println("Accessing: " + NO); 17 Thread.sleep((long) (10000); 18 semp.release(); 19 } catch (InterruptedException e) { 20 } 21 } 22 }; 23 exec.execute(run); 24 } 25 exec.shutdown(); 26 }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

2、CountDownLatch闭锁：允许一个或多个线程一直等待，直到其他线程的操作执行完后再执行。CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

主要方法： 
1. CountDownLatch.await()：将某个线程阻塞住，直到计数器count=0才恢复执行。 
2. CountDownLatch.countDown()：将计数器count减1。

使用场景： 
1. 实现最大的并行性：有时我们想同时启动多个线程，实现最大程度的并行性。例如，我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch，并让所有线程都在这个锁上等待，那么我们可以很轻松地完成测试。我们只需调用 一次countDown()方法就可以让所有的等待线程同时恢复执行。 
2. 开始执行前等待n个线程完成各自任务：例如应用程序启动类要确保在处理用户请求前，所有N个外部系统已经启动和运行了。 
3. 死锁检测：一个非常方便的使用场景是，你可以使用n个线程访问共享资源，在每次测试阶段的线程数目是不同的，并尝试产生死锁。 
4. 计算并发执行某个任务的耗时。

示例代码：

public class CountDownLatchTest { public void timeTasks(int nThreads, final Runnable task) throws InterruptedException{ final CountDownLatch startGate = new CountDownLatch(1); final CountDownLatch endGate = new CountDownLatch(nThreads); for(int i = 0; i < nThreads; i++){ Thread t = new Thread(){ public void run(){ try{ startGate.await(); try{ task.run(); }finally{ endGate.countDown(); } }catch(InterruptedException ignored){ } } }; t.start(); } long start = System.nanoTime(); System.out.println("打开闭锁"); startGate.countDown(); endGate.await(); long end = System.nanoTime(); System.out.println("闭锁退出，共耗时" + (end-start)); } public static void main(String[] args) throws InterruptedException{ CountDownLatchTest test = new CountDownLatchTest(); test.timeTasks(5, test.new RunnableTask()); } class RunnableTask implements Runnable{ @Override public void run() { System.out.println("当前线程为：" + Thread.currentThread().getName()); } } 执行结果为： 打开闭锁 当前线程为：Thread-0 当前线程为：Thread-3 当前线程为：Thread-2 当前线程为：Thread-4 当前线程为：Thread-1 闭锁退出，共耗时1109195
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55

3、CyclicBarrier栅栏：用于阻塞一组线程直到某个事件发生。所有线程必须同时到达栅栏位置才能继续执行下一步操作，且能够被重置以达到重复利用。而闭锁是一次性对象，一旦进入终止状态，就不能被重置。

示例代码：

public class CyclicBarrierTest { private final CyclicBarrier barrier; private final Worker[] workers; public CyclicBarrierTest(){ int count = Runtime.getRuntime().availableProcessors(); this.barrier = new CyclicBarrier(count, new Runnable(){ @Override public void run() { System.out.println("所有线程均到达栅栏位置，开始下一轮计算"); } }); this.workers = new Worker[count]; for(int i = 0; i< count;i++){ workers[i] = new Worker(i); } } private class Worker implements Runnable{ int i; public Worker(int i){ this.i = i; } @Override public void run() { for(int index = 1; index < 3;index++){ System.out.println("线程" + i + "第" + index + "次到达栅栏位置，等待其他线程到达"); try { //注意是await,而不是wait barrier.await(); } catch (InterruptedException e) { e.printStackTrace(); return; } catch (BrokenBarrierException e) { e.printStackTrace(); return; } } } } public void start(){ for(int i=0;i<workers.length;i++){ new Thread(workers[i]).start(); } } public static void main(String[] args){ new CyclicBarrierTest().start(); } } 执行结果为： 线程0第1次到达栅栏位置，等待其他线程到达 线程1第1次到达栅栏位置，等待其他线程到达 线程2第1次到达栅栏位置，等待其他线程到达 线程3第1次到达栅栏位置，等待其他线程到达 所有线程均到达栅栏位置，开始下一轮计算 线程3第2次到达栅栏位置，等待其他线程到达 线程2第2次到达栅栏位置，等待其他线程到达 线程0第2次到达栅栏位置，等待其他线程到达 线程1第2次到达栅栏位置，等待其他线程到达 所有线程
标签：

相关文章