CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS

时间  2020-12-27
栏目 Java 繁體版
原文   原文链接

学习自http://www.importnew.com/21889.html

https://blog.csdn.net/yanyan19880509/article/details/52349056

https://www.cnblogs.com/waterystone/p/4920797.html

1
2
3
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

下面看一个例子大家就清楚CountDownLatch的用法了：

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
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();
         }
      }
}

执行结果：

1
2
3
4
5
6
7
线程Thread- 0 正在执行
线程Thread- 1 正在执行
等待 2 个子线程执行完毕...
线程Thread- 0 执行完毕
线程Thread- 1 执行完毕
2 个子线程已经执行完毕
继续执行主线程


在new CountDownLatch(2)的时候


对于 CountDownLatch 来说，state=2表示所有调用await方法的线程都应该阻塞，等到同一个latch被调用两次countDown后才能唤醒沉睡的线程

为了充分了解AQS里的链表，这里假设上面挂起等待的线程数为2个


当latch被成功减到0后，AQS的state就成了0。那个成功减到0的那个线程。然后节点3被唤醒了。当节点3醒来后，发现自己是通知状态，然后删除自己，唤醒节点4。

上面的流程，如果落实到代码，把 state置为0的那个线程，会判断head指向节点的状态，如果为通知状态，则唤醒后续节点，即线程3节点，然后head指向线程3节点，head指向的旧节点会被删除掉。当线程3恢复执行后，发现自身为通知状态，又会把head指向线程4节点，然后删除自身节点，并唤醒线程4。

线程节点的状态是什么时候设置上去的？其实，一个线程在阻塞之前，就会把它前面的节点设置为通知状态，这样便可以实现链式唤醒机制了。


AQS是一些同步的抽象，简单介绍一下


两大元素

1.volatile int state->共享资源

2.FIFO线程等待队列，多线程争用资源被阻塞时会进入此队列


AQS定义两种资源占用方式：

1.Exclusive（独占，只有一个线程能执行，如ReentrantLock）

2.Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）

自定义同步器在实现时只需要实现state的增减即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在底层实现好了


自定义同步器的时候，主要实现以下几种方法：

1.isHeldExclusively()，该线程是否正在独占资源，只有用到condition的时候才需要去实现这个方法

2.tryAcquire(int)，独占。尝试获取资源，成功返回true，失败返回fasle

3.tryRelease(int)

4.tryAcquireShared(int)，共享。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。

5.tryReleaseShared(int)


以ReentrantLock为例

state初始化为0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到A线程unlock()到state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A线程自己是可以重复获取此锁的（state会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证state是能回到零态的。


以CountDownLatch以例

任务分为N个子线程去执行，state也初始化为N（注意N要与线程个数一致）。这N个子线程是并行执行的，每个子线程执行完后countDown()一次，state会CAS减1。等到所有子线程都执行完后(即state=0)，会unpark()主调用线程，然后主调用线程就会从await()函数返回，继续后余动作。


深入看一下CountDownLatch

首先是他的内部类

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    Sync(int count) {
        setState(count);
    }

    int getCount() {
        return getState();
    }

    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    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;
        }
    }
}

private final Sync sync;

tryAcquireShared不知道有啥用，理论上应该是来获取资源的，但是对于CountDownLatch不需要了，因为他只需要一开始设定state即可。不过CountDownLatch为什么是共享锁我理解了，他允许多个线程同时使用CPU资源。


Node.waitStatus

-CANCELLED：值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
-SIGNAL：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
-CONDITION：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
-PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
0状态：值为0，代表初始化状态。

AQS在判断状态时，通过用waitStatus>0表示取消状态，而waitStatus<0表示有效状态。


CountDownLatch的-1方法

public void countDown() {
    sync.releaseShared(1);
}

AQS中

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

前面讲到：releaseShared就是让state-1，当你减到0的时候，减到0的那个线程开始唤醒线程。

这里和代码逻辑对应一下。

tryReleaseShared就是让state-1。如果减到0，就返回true了，开始执行doReleaseShared，唤醒线程。

private void doReleaseShared() {
    for (;;) {
        Node h = head;//头结点
        if (h != null && h != tail) {//如果不为空，且不是尾节点
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {//如果是待唤醒，waitStatus改为初始化状态
                if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);//找到他后面的第一个节点，也就是允许使用cpu，同时解锁了
            }
            else if (ws == 0 &&//如果是初始化状态 希望把他变成可运行状态
                     !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)//如果head变了，就继续loop                   // loop if head changed
            break;
    }
}

所以doReleaseShared的整体意思是：唤醒下一个线程

unparkSuccessor

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);//如果是可运行、可唤醒，那就置为初始化状态

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;//如果为空或者cancel
        for (Node p = tail; p != node && p != null; p = p.prev)//从后往前找
            if (p.waitStatus <= 0)//目的是为了找到一个可运行、可唤醒的
                s = p;
    }
    if (s != null)//如果找到了
        LockSupport.unpark(s.thread);//解锁这个线程
}

整体的意思就是，找到他后面的第一个节点，也就是允许使用cpu，同时解锁了


那么被唤醒的线程是在哪里唤醒下一个线程的呢？

需要从await入手

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)//如果state!=0，之前初始化已经让state=2了
        doAcquireSharedInterruptibly(arg);
}


private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);//尾部加一个node
    try {
        for (;;) {
            final Node p = node.predecessor();//前一个node
            if (p == head) {//是head
                int r = tryAcquireShared(arg);
                if (r >= 0) {//state==0
                    setHeadAndPropagate(node, r);//那就开始，并唤醒下一个线程
                    p.next = null; // help GC
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())//锁住咯，并让上一个线程置为通知状态
                throw new InterruptedException();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}

CountDownLatch就分析的差不多了


为何CountDownLatch被称为共享锁？

因为CountDownLatch latch = new CountDownLatch(2)，他允许了多个线程同时使用CPU资源，而ReentrantLock是单行道。在这个实例化中，锁可以认为有2道（state），每个线程执行完后，通过cas方式正确地减了一道。


再讲一下CyclicBarrier、 Semaphore，不过就不扯源码了


CyclicBarrier

可以实现让一组线程等待至某个状态之后再全部同时执行

例子

续

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
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( "所有线程写入完毕，继续处理其他任务..." );
         }
     }
}

执行结果：

1
2
3
4
5
6
7
8
9
10
11
12
线程Thread- 0 正在写入数据...
线程Thread- 3 正在写入数据...
线程Thread- 2 正在写入数据...
线程Thread- 1 正在写入数据...
线程Thread- 2 写入数据完毕，等待其他线程写入完毕
线程Thread- 0 写入数据完毕，等待其他线程写入完毕
线程Thread- 3 写入数据完毕，等待其他线程写入完毕
线程Thread- 1 写入数据完毕，等待其他线程写入完毕
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...


总结CountDownLatch、CyclicBarrie的区别

前者是手动-1，另外线程先阻塞着。state减到0后，另外线程可以跑了

后者是若干线程先阻塞着，等到阻塞到若干数量后，所有线程就可以跑了


Semaphore

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

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
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();
             }
         }
     }
}

执行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
工人 0 占用一个机器在生产...
工人 1 占用一个机器在生产...
工人 2 占用一个机器在生产...
工人 8
9
10
11
12
13
14
15
16
工人 0 占用一个机器在生产...
工人 1 占用一个机器在生产...
工人 2 占用一个机器在生产...
工人 4 占用一个机器在生产...
工人 5 占用一个机器在生产...
工人 0 释放出机器
8
9
10
11
12
13
14
15
16
工人 0 占用一个机器在生产...
工人 1 占用一个机器在生产...
工人 2 占用一个机器在生产...
工人 4 占用一个机器在生产...
工人 5 占用一个机器在生产...
工人 0 释放出机器
工人 2 释放出机器
工人 3 占用一个机器在生产...
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程