天天看点

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

一、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),允许多个线程获得许可并执行。可以用来控制同时访问某个特定资源的操作数量,或者同时执行某个指定操作的数量。

示例代码:

public class TIJ_semaphore {
      public static void main(String[] args) {
          ExecutorService exec = Executors.newCachedThreadPool();
          final Semaphore semp = new Semaphore(); // 5 permits
  
         for (int index = ; index < ; index++) {
             final int NO = index;
             Runnable run = new Runnable() {
                 public void run() {
                     try {
                           // if 1 permit avaliable, thread will get a permits and go; if no permit avaliable, thread will block until 1 avaliable
                         semp.acquire(); 
                         System.out.println("Accessing: " + NO);
                         Thread.sleep((long) ();
                         semp.release();
                     } catch (InterruptedException e) {
                     }
                 }
             };
             exec.execute(run);
         }
         exec.shutdown();
     }
           

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();  
        final CountDownLatch endGate = new CountDownLatch(nThreads);  

        for(int i = ; 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(, test.new RunnableTask());  
    }  

    class RunnableTask implements Runnable{  

        @Override  
        public void run() {  
            System.out.println("当前线程为:" + Thread.currentThread().getName());  

        }     
    }  

执行结果为:  
打开闭锁  
当前线程为:Thread-  
当前线程为:Thread-  
当前线程为:Thread-  
当前线程为:Thread-  
当前线程为:Thread-  
闭锁退出,共耗时  
           

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 = ; 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 = ; index < ;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=;i<workers.length;i++){
            new Thread(workers[i]).start();
        }
    }

    public static void main(String[] args){
        new CyclicBarrierTest().start();
    }
}

执行结果为:  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
所有线程均到达栅栏位置,开始下一轮计算  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
线程第次到达栅栏位置,等待其他线程到达  
所有线程均到达栅栏位置,开始下一轮计算 
           

继续阅读