黑马程序员技术交流社区

标题: 【上海校区】Java中的显示锁ReentrantLock使用与AbstractQueuedSync... [打印本页]
作者: 不二晨    时间: 2018-11-23 10:07
标题: 【上海校区】Java中的显示锁ReentrantLock使用与AbstractQueuedSync...
考虑一个场景，轮流打印0-100以内的技术和偶数。通过使用 synchronize 的 wait，notify机制就可以实现，核心思路如下：
使用两个线程，一个打印奇数，一个打印偶数。这两个线程会共享一个数据，数据每次自增，当打印奇数的线程发现当前要打印的数字不是奇数时，执行等待，否则打印奇数，并将数字自增1，对于打印偶数的线程也是如此
//打印奇数的线程private static class OldRunner implements Runnable{    private MyNumber n;    public OldRunner(MyNumber n) {        this.n = n;    }    public void run() {        while (true){            n.waitToOld();  //等待数据变成奇数            System.out.println("old:" + n.getVal());            n.increase();            if (n.getVal()>98){                break;            }        }    }}//打印偶数的线程private static class EvenRunner implements Runnable{    private MyNumber n;    public EvenRunner(MyNumber n) {        this.n = n;    }    public void run() {        while (true){            n.waitToEven();            //等待数据变成偶数            System.out.println("even:"+n.getVal());            n.increase();             if (n.getVal()>99){                break;            }        }    }}复制代码共享的数据如下
private static class MyNumber{    private int val;    public MyNumber(int val) {        this.val = val;    }    public int getVal() {        return val;    }    public synchronized void increase(){        val++;        notify(); //数据变了，唤醒另外的线程    }    public synchronized void  waitToOld(){        while ((val % 2)==0){            try {                System.out.println("i am "+Thread.currentThread().getName()+" ,but now is even:"+val+",so wait");                wait(); //只要是偶数，一直等待            } catch (InterruptedException e) {                e.printStackTrace();            }        }    }    public synchronized void waitToEven(){        while ((val % 2)!=0){            try {                System.out.println("i am "+Thread.currentThread().getName()+"  ,but now old:"+val+",so wait");                wait(); //只要是奇数，一直等待            } catch (InterruptedException e) {                e.printStackTrace();            }        }    }}复制代码运行代码如下
MyNumber n = new MyNumber(0);Thread old=new Thread(new OldRunner(n),"old-thread");Thread even = new Thread(new EvenRunner(n),"even-thread");old.start();even.start();复制代码运行结果如下
i am old-thread ,but now is even:0,so waiteven:0i am even-thread  ,but now old:1,so waitold:1i am old-thread ,but now is even:2,so waiteven:2i am even-thread  ,but now old:3,so waitold:3i am old-thread ,but now is even:4,so waiteven:4i am even-thread  ,but now old:5,so waitold:5i am old-thread ,but now is even:6,so waiteven:6i am even-thread  ,but now old:7,so waitold:7i am old-thread ,but now is even:8,so waiteven:8复制代码上述方法使用的是 synchronize的 wait notify机制，同样可以使用显示锁来实现，两个打印的线程还是同一个线程，只是使用的是显示锁来控制等待事件
private static class MyNumber{    private Lock lock = new ReentrantLock();    private Condition condition = lock.newCondition();    private int val;    public MyNumber(int val) {        this.val = val;    }    public int getVal() {        return val;    }    public  void increase(){        lock.lock();        try {            val++;            condition.signalAll(); //通知线程        }finally {            lock.unlock();        }    }    public  void  waitToOld(){        lock.lock();        try{            while ((val % 2)==0){                try {                    System.out.println("i am should print old ,but now is even:"+val+",so wait");                    condition.await();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }finally {            lock.unlock();        }    }    public void waitToEven(){        lock.lock(); //显示的锁定        try{            while ((val % 2)!=0){                try {                    System.out.println("i am should print even ,but now old:"+val+",so wait");                    condition.await();//执行等待                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }finally {            lock.unlock(); //显示的释放        }    }}复制代码同样可以得到上述的效果
显示锁的功能显示锁在java中通过接口Lock提供如下功能

Condition的功能接口Condition把Object的监视器方法wait和notify分离出来，使得一个对象可以有多个等待的条件来执行等待，配合Lock的newCondition来实现。
ReentrantLock从源码中可以看到，ReentrantLock的所有实现全都依赖于内部类Sync和ConditionObject。
Sync本身是个抽象类，负责手动lock和unlock,ConditionObject则实现在父类AbstractOwnableSynchronizer中，负责await与signalSync的继承结构如下

Sync的两个实现类，公平锁和非公平锁
公平的锁会把权限给等待时间最长的线程来执行,非公平则获取执行权限的线程与线程本身的等待时间无关
默认初始化ReentrantLock使用的是非公平锁，当然可以通过指定参数来使用公平锁
public ReentrantLock() {   sync = new NonfairSync();}复制代码
当执行获取锁时，实际就是去执行 Sync 的lock操作：
public void lock() {    sync.lock();}复制代码对应在不同的锁机制中有不同的实现
二者都调用父类AbstractQueuedSynchronizer的方法
public final void acquire(int arg) {    if (!tryAcquire(arg) &&        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //一旦抢失败，就会进入队列，进入队列后则是依据FIFO的原则来执行唤醒        selfInterrupt();}复制代码当执行unlock时，对应方法在父类AbstractQueuedSynchronizer中
public final boolean release(int arg) {    if (tryRelease(arg)) {        Node h = head;        if (h != null && h.waitStatus != 0)            unparkSuccessor(h);        return true;    }    return false;}复制代码公平锁和非公平锁则分别对获取锁的方式tryAcquire 做了实现，而tryRelease的实现机制则都是一样的
公平锁实现tryAcquire源码如下
protected final boolean tryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState(); //获取当前的同步状态    if (c == 0) {        //等于0 表示没有被其它线程获取过锁        if (!hasQueuedPredecessors() &&            compareAndSetState(0, acquires)) {            //hasQueuedPredecessors 判断在当前线程的前面是不是还有其它的线程，如果有，也就是锁sync上有一个等待的线程，那么它不能获取锁，这意味着，只有等待时间最长的线程能够获取锁，这就是是公平性的体现            //compareAndSetState 看当前在内存中存储的值是不是真的是0，如果是0就设置成accquires的取值。对于JAVA，这种需要直接操作内存的操作是通过unsafe来完成，具体的实现机制则依赖于操作系统。            //存储获取当前锁的线程            setExclusiveOwnerThread(current);            return true;        }    }    else if (current == getExclusiveOwnerThread()) {        //判断是不是当前线程获取的锁        int nextc = c + acquires;        if (nextc < 0)//一个线程能够获取同一个锁的次数是有限制的，就是int的最大值            throw new Error("Maximum lock count exceeded");        setState(nextc); //在当前的基础上再增加一次锁被持有的次数        return true;    }    //锁被其它线程持有，获取失败    return false;}复制代码非公平锁实现tryAcquire获取的关键实现为nonfairTryAcquire,源码如下
final boolean nonfairTryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState();    if (c == 0) {        //锁没有被持有        //可以看到这里会无视sync queue中是否有其它线程，只要执行到了当前线程，就会去获取锁        if (compareAndSetState(0, acquires)) {             setExclusiveOwnerThread(current); //在判断一次是不是锁没有被占有，没有就去标记当前线程拥有这个锁了            return true;        }    }    else if (current == getExclusiveOwnerThread()) {        int nextc = c + acquires;         if (nextc < 0) // overflow                        throw new Error("Maximum lock count exceeded");        setState(nextc);//如果当前线程已经占有过，增加占有的次数        return true;    }    return false;}复制代码释放锁的机制protected final boolean tryRelease(int releases) {    int c = getState() - releases;    if (Thread.currentThread() != getExclusiveOwnerThread()) //只能是线程拥有这释放        throw new IllegalMonitorStateException();    boolean free = false;    if (c == 0) {        //当占有次数为0的时候，就认为所有的锁都释放完毕了        free = true;         setExclusiveOwnerThread(null);    }    setState(c); //更新锁的状态    return free;}复制代码从源码的实现可以看到
ReentrantLock的tryLock()与tryLock(long timeout, TimeUnit unit):
public boolean tryLock() {//本质上就是执行一次非公平的抢锁   return sync.nonfairTryAcquire(1); }复制代码有时限的tryLock核心代码是 sync.tryAcquireNanos(1, unit.toNanos(timeout));，由于有超时时间，它会直接放到等待队列中，他与后面要讲的AQS的lock原理中acquireQueued的区别在于park的时间是有限的，详见源码 AbstractQueuedSynchronizer.doAcquireNanos
AbstractQueuedSynchronizer无论是公平锁还是非公平锁，它们的实现都依赖于AbstractQueuedSynchronizer，它提供了一个基于先进先出等待队列 实现block locks和synchronizers的框架。特性如下
lock()原理当ReentrantLock获取锁失败时，会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
private Node addWaiter(Node mode) {     Node node = new Node(Thread.currentThread(), mode); //创建一个节点，存储当前的线程，以及锁持有的模式，对于 ReentrantLock来说就是 独占 型    // Try the fast path of enq; backup to full enq on failure        Node pred = tail;    if (pred != null) {        node.prev = pred;        if (compareAndSetTail(pred, node)) {//CAS操作，如果当前的尾部节点没有被其它线程更改，那么把新的节点设置成队列的尾部            pred.next = node;            return node;        }    }    enq(node);//首次入队    return node;}复制代码获取失败进行入队操作，首先就是往队列中添加一个正在等待的节点Node

从Node本身的结构可以看到，AQS(AbstractQueuedSynchronizer)本身就维护了一个双向链表，用来存放等待中的线程。链表的每个节点，代表那个线程，是独占还是共享锁。
创建好节点之后，便执行入队操作，对于首次创建队列
private Node enq(final Node node) {    for (;;) {       //借助CAS机制实现无锁操作，所以需要一直执行直到CAS成功        Node t = tail;        if (t == null) { // 初始化发生在第一次创建队列，这样的好处是，当竞争不激烈的时候，实际上也就不会发生这些操作，性能也会好些             if (compareAndSetHead(new Node()))                tail = head;        } else {            node.prev = t;            if (compareAndSetTail(t, node)) {                t.next = node;                return t;            }        }    }}复制代码可以看到，入队也就是从队尾插入新的等待线程，入队完毕，也就开始去进行不断的尝试，直到获取锁成功,可以看到，对于lock来说，其实已经是阻塞了
final boolean acquireQueued(final Node node, int arg) {    boolean failed = true;    try {        boolean interrupted = false;        for (;;) {            final Node p = node.predecessor(); //优先执行当前节点的前一个节点            if (p == head && tryAcquire(arg)) { //仅当当前节点的前一个节点是head,才去获取线程，这里可以看出其实先等待的线程是会优先处理，也就是FIFO原则                setHead(node);                 p.next = null; // help GC    ,释放掉当前线程在队列中的引用，也可以看做’出队'了                            failed = false;                //执行到这里说明获取锁成功                return interrupted;            }            //执行到这里说明存在竞争，有多个线程都在等待一个锁            if (shouldParkAfterFailedAcquire(p, node) &&                parkAndCheckInterrupt()) //里面会对当前线程执行中断，当被唤醒时，继续循环                    //如果线程被中断，设置中断标记，区别于 doAcquireInterruptibly，doAcquireInterruptibly是直接抛出异常，这也就是 lockInterruptibly能够抛出中断的原因                    interrupted = true;        }    } finally {        if (failed)            cancelAcquire(node);    }}复制代码
从这里可以看到，无论锁是公平锁还是非公平锁，只要被放入了等待队列，此时的执行依然是谁先等待就先执行谁  ,非公平锁体现在新来的线程会无视已经等了的线程，可以优先去抢锁,所以公平体现在第一次参与抢锁的线程会去等待已经在等待队列中的线程,非公平并不是说从已经在等待的线程队列里面随便选一个
shouldParkAfterFailedAcquire的源码如下
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {    int ws = pred.waitStatus; //查看前一个节点的等待状态    if (ws == Node.SIGNAL)            //已经尝试过获取锁，可以执行park了            return true;    if (ws > 0) {           do {            //去掉队列中所有已经取消的线程            node.prev = pred = pred.prev;        } while (pred.waitStatus > 0);        pred.next = node;    } else {                     //此时当前线程的前一个节点的等待状态必定是0或者PROGATE，这表明当前线程在park之前可以再尝试一次去获取锁,也就是说前一个节点可能刚获取到SIGNAL        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);    }    return false;}复制代码waitStatus：等待的状态，共有5种
parkAndCheckInterrupt主要是park当前线程
private final boolean parkAndCheckInterrupt() {    //当获取不到许可时，阻塞线程，解除阻塞状态的情况如下：    //1 某个线程对这个线程调用了unpark方法    //2 某个线程中断了这个线程    //3 这个方法毫无理由的返回了 [park比较奇特的地方],基于这样，调用的时候必须去判断park的条件，以及当它返回的时候，去设置中断的状态    LockSupport.park(this);     //返回线程的中断状态    return Thread.interrupted();}复制代码至此lock()执行结束
unlock()原理当执行unlock时，ReentrentLock执行对应的Release
public final boolean release(int arg) {    if (tryRelease(arg)) {        //执行这里表示所已经被释放,可以让它的下一个节点来抢锁        Node h = head;        if (h != null && h.waitStatus != 0)            unparkSuccessor(h); //h.waitStatus == 0 表示还没有执行park,自然不需要unpark        return true;    }    return false;}复制代码如果release成功，即当前线程持有的所有锁都已经释放，那么就可以执行 unparkSuccessor,从源码可以看到，unpark是从头部开始进行的，结合lock的原理，可知AQS本身就是一个先进先出的队列unparkSuccessor源码如下
private void unparkSuccessor(Node node) {     int ws = node.waitStatus;    if (ws < 0)        compareAndSetWaitStatus(node, ws, 0);     Node s = node.next; //有可能线程被取消了，所以从尾部往前遍历，到最近一个没有被取消的线程    if (s == null || s.waitStatus > 0) {        s = null;        for (Node t = tail; t != null && t != node; t = t.prev)            if (t.waitStatus <= 0)                s = t; //确保线程没有取消    }    if (s != null)        LockSupport.unpark(s.thread); //恢复线程}复制代码至此unlock()完毕
await的原理public final void await() throws InterruptedException {    if (Thread.interrupted())        throw new InterruptedException();//当前线程已经中断了，抛出中断异常    //添加一个新的waiter到condition queue中,这个新的Node的waitStatus会被标记为CONDITION    Node node = addConditionWaiter();     //释放当前线程拥有的锁,即从sync queue中去掉当前线程    int savedState = fullyRelease(node);    int interruptMode = 0;    while (!isOnSyncQueue(node)) {    //如果当前线程不在持有锁的队列里头,对他进行休眠，当其它线程执行 unlock的时候，释放锁，就会执行unpark操作，此时它会被唤醒，唤醒后，如果它在syn队列里头，开始继续往下执行。(这个插入操作则是由signal完成)        LockSupport.park(this);         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)            break;//等待的过程中线程中断了，退出    }//重新竞争锁,相当于执行了lock操作    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)    //再次去获取锁，如果当前的线程在park的时候是被中断了，并且ConditionObject并不是由于中断返回，这里再次标记为中断        interruptMode = REINTERRUPT;      if (node.nextWaiter != null)     //清除非Condition模式的线程,而在signal中有先关操作将conditon的线程设置成非condition         unlinkCancelledWaiters();    if (interruptMode != 0)    //上报等待的过程中发生了中断,如果是要抛出中断，就抛出，否则再次执行中断        reportInterruptAfterWait(interruptMode); }复制代码isOnSyncQueue源码如下
final boolean isOnSyncQueue(Node node) {    if (node.waitStatus == Node.CONDITION || node.prev == null)    //node本身是调用了 await 方法,或者没有在获取锁的队列里头,[如果在里头必定有一个前置的节点]        return false;     if (node.next != null)     //当前节点存在下一个节点，那么它肯定是执行过 enq ,即获取过锁        return true;     // CAS失败的时候，有可能 node.rev是没有的，因此需要从头到尾遍历一次   return findNodeFromTail(node); }复制代码checkInterruptWhileWaiting源码如下
private int checkInterruptWhileWaiting(Node node) {    return Thread.interrupted() ?        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :        0;}final boolean transferAfterCancelledWait(Node node) {    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {    //线程中断重新获取锁,并且设置waitStatus为0，以便后续线程从condition queue清除        enq(node);         return true;    }    while (!isOnSyncQueue(node))    //如果CAS失败，只要当前节点没有在Sync queue中，那么一直自旋，每次都会交出执行权限        Thread.yield();     return false;}复制代码可以看到，await其实就是释放线程原有的锁，并把它放入conditon队列中，然后执行阻塞。等唤醒的时候，重新获取锁，并清掉condition queue中的线程。 至此await执行结束
singnal的原理public final void signal() {    if (!isHeldExclusively()) //只有当前线程持有了锁，才能释放        throw new IllegalMonitorStateException();    Node first = firstWaiter;    if (first != null)        doSignal(first);//优先释放队列头的,也就是等待时间最长的condition node}private void doSignal(Node first) {    do {        if ( (firstWaiter = first.nextWaiter) == null)            lastWaiter = null;        first.nextWaiter = null;    } while (!transferForSignal(first) &&             (first = firstWaiter) != null);}//将节点从condition queue转移到sync queuefinal boolean transferForSignal(Node node) {    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))        return false; //设置为非等待失败，则不继续转移//CAS设置等待状态为0成功Node p = enq(node); //新节点放入sync queue，并返回原来的尾部节点，也就是新节点的前一个节点    int ws = p.waitStatus;    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) //参考shouldParkAfterFailedAcquire        LockSupport.unpark(node.thread);//如果当前节点的前一个节点线程已经取消,或者将当前节点的前一个节点线程的waitStatus设置成SIGNAL失败,则直接唤醒当前线程    return true;}复制代码可以看到signal最关键的信息就是去掉等待队列中的CONDITION状态，并将线程加入sync队列，至此signal结束


【转载】
作者：爬蜥
链接：https://juejin.im/post/5bf18909f265da611036a4f0



作者: 小影姐姐    时间: 2018-11-26 14:13
作者: 不二晨    时间: 2018-11-28 15:43
奈斯



欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/) 黑马程序员IT技术论坛 X3.2