【java并发】juc高级锁机制探讨

程序员文章站 2022-06-08 20:31:11

...

最近在看一些juc相关的设计和源码，接上文 ：【java并发】基于JUC CAS原理,自己实现简单独占锁

本文探讨一下juc里面提供的一些高级锁机制和基本原理。

JUC高级锁机制简介

Juc 提供了高级锁的一些特性和应用，如：

ReentrantLock ：和 synchronized 具有差不多的语义，独占锁，同时只有一个线程能获得锁。

ReentrantReadWriteLock ：读写锁。读允许共享，写独占。适用于读频繁的场景。

CountDownLatch : 闭锁。使用场景类似比赛鸣枪，在没有鸣枪之前所有的运动员 ( 线程 ) 都必须等待。说白了就是使用于一个或多个线程等待某一个条件成立了，触发运行。

Semaphore ：信号量。使用场景类似通行证，通行证数量有限，拿到证的才能通行。

CyclicBarrier : 周期障碍。语义上和 CountDownLatch 有点类似，只有几个线程打到一个共同的状态之后，触发后续动作继续。不同在于 1. 达到共同状态后，可以指定一个后续触发的线程对象。 2. 周期性意味着可以周期运行。

FutureTask ：带有返回值的异步执行。

至于各种锁机制使用场景这里不赘述，后面还有机会加一些例子。

以上不同的锁机制和使用场景，不管我们我们叫锁、闭锁、信号量等等。抽象之后都有一种共同的语义：

多线程并发的执行，之间通过某种共享状态来同步，只有当状态满足 xxxx 条件，才能触发线程执行 xxxx 。

这个共同的语义可以称之为同步器。可以认为以上所有的锁机制都可以基于同步器定制来实现的。

如果要实现一个特定场景的锁来同步线程的执行，其实并不难，如上文 【java并发】基于JUC CAS原理,自己实现简单独占锁 。而juc里的思想是将这些场景抽象出来的语义通过统一的同步框架来支持。

juc 里所有的这些锁机制都是基于 AQS （ AbstractQueuedSynchronizer ）框架上构建的。下面简单介绍下 AQS （ AbstractQueuedSynchronizer ）。可以参考Doug Lea的论文The java.util.concurrent Synchronizer Framework

AQS框架

AbstractQueuedSynchronizer 是一个抽象类，里面定义了同步器的基本框架，实现了基本的结构功能。只留有状态条件的维护由具体同步器根据具体场景来定制，如上面提到的 ReentrantLock 、 RetrantReadWriteLock和CountDownLatch 等等。

一个同步器至少需要包含两个功能：

1. 获取同步状态

如果允许，则获取锁，如果不允许就阻塞线程，直到同步状态允许获取。

2. 释放同步状态

修改同步状态，并且唤醒等待线程。

根据作者论文， aqs 同步机制同时考虑了如下需求：

1. 独占锁和共享锁两种机制。

2. 线程阻塞后，如果需要取消，需要支持中断。

3. 线程阻塞后，如果有超时要求，应该支持超时后中断的机制。

实现涉及基本技术原理

1. 状态位

提供 volatile 变量 state; 用于同步线程之间的共享状态。通过 CAS 和 volatile 保证其原子性和可见性。对应源码里的定义：

    /**
     * 同步状态
     */
    private volatile int state;

    /**
     *cas
     */
    protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

2. 线程阻塞和唤醒

有别于wait和notiry。这里利用 jdk1.5 开始提供的 LockSupport.park() 和 LockSupport.unpark() 的本地方法实现，实现线程的阻塞和唤醒。

3. 阻塞线程节点队列 CHL Node queue 。

根据论文里描述， AQS 里将阻塞线程封装到一个内部类 Node 里。并维护一个 CHL Node FIFO 队列。 CHL 队列是一个非阻塞的 FIFO 队列，也就是说往里面插入或移除一个节点的时候，在并发条件下不会阻塞，而是通过自旋锁和 CAS 保证节点插入和移除的原子性。实现无锁且快速的插入。关于非阻塞算法可以参考 Java 理论与实践: 非阻塞算法简介。CHL队列对应代码如下：

     /**
     * CHL头节点
     */ 
   private transient volatile Node head;
    /**
     * CHL尾节点
     */
    private transient volatile Node tail;

Node节点是对Thread的一个封装，结构大概如下：

    static final class Node {
        /** 代表线程已经被取消*/
        static final int CANCELLED =  1;
        /** 代表后续节点需要唤醒 */
        static final int SIGNAL    = -1;
        /** 代表线程在等待某一条件/
        static final int CONDITION = -2;
        /** 标记是共享模式*/
        static final Node SHARED = new Node();
        /** 标记是独占模式*/
        static final Node EXCLUSIVE = null;

        /**
         * 状态位 ，分别可以使CANCELLED、SINGNAL、CONDITION、0
         */
        volatile int waitStatus;

        /**
         * 前置节点
         */
        volatile Node prev;

        /**
         * 后续节点
         */
        volatile Node next;

        /**
         * 节点代表的线程
         */
        volatile Thread thread;

        /**
         *连接到等待condition的下一个节点
         */
        Node nextWaiter;

    }

AQS 源码

AQS实现了一个同步器的基本结构，下面以独占锁和非独占锁区分来看看 AQS 的几个主要方法：

独占模式

独占获取： tryAcquire 本身不会阻塞线程，如果返回 true 成功就继续，如果返回 false 那么就阻塞线程并加入阻塞队列。

    public final void acquire(int arg) {

        if (!tryAcquire(arg) &&

            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取失败，则加入等待队列

            selfInterrupt();

}

独占且可中断模式获取：支持中断取消

public final void acquireInterruptibly(int arg) throws InterruptedException {

        if (Thread.interrupted())
            throw new InterruptedException();
        if (!tryAcquire(arg))

            doAcquireInterruptibly(arg);

    }

独占且支持超时模式获取：带有超时时间，如果经过超时时间则会退出。

    public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {

         if (Thread.interrupted())

             throw new InterruptedException();

         return tryAcquire(arg) ||

             doAcquireNanos(arg, nanosTimeout);

}

独占模式释放：释放成功会唤醒后续节点

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

共享模式

共享模式获取

    public final void acquireShared(int arg) {

        if (tryAcquireShared(arg) < 0)

            doAcquireShared(arg);

}

可中断模式共享获取

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

共享模式带定时获取

    public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
         if (Thread.interrupted())
             throw new InterruptedException();
         return tryAcquireShared(arg) >= 0 ||
             doAcquireSharedNanos(arg, nanosTimeout);
    }

共享锁释放

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

注意以上框架只定义了一个同步器的基本结构框架，的基本方法里依赖的 tryAcquire 、 tryRelease 、 tryAcquireShared 、 tryReleaseShared 四个方法在 AQS 里没有实现，这四个方法不会涉及线程阻塞，而是由各自不同的使用场景根据情况来定制:

    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }
    protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();

    }
    protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }

从以上源码可以看出AQS实现基本的功能：

AQS虽然实现了acquire，和release方法是可能阻塞的，但是里面调用的tryAcquire和tryRelease是由子类来定制的且是不阻塞的可。以认为同步状态的维护、获取、释放动作是由子类实现的功能，而动作成功与否的后续行为时有AQS框架来实现。所以可以认为同步器实现了一下功能：

1.同步器基本范式、结构

2.状态获取、释放成功或失败的后续行为，如线程的阻塞、唤醒机制

3.线程阻塞队列的维护

状态获取、释放动作本身是由子类来定义的。

还有以下一些私有方法，用于辅助完成以上的功能：

final boolean acquireQueued(final Node node, int arg) ：申请队列

private Node enq(final Node node) : 入队

private Node addWaiter(Node mode) ：以mode创建创建节点，并加入到队列

private void unparkSuccessor(Node node) ：唤醒节点的后续节点，如果存在的话。

private void doReleaseShared() ：释放共享锁

private void setHeadAndPropagate(Node node, int propagate)：设置头，并且如果是共享模式且propagate大于0，则唤醒后续节点。

private void cancelAcquire(Node node) ：取消正在获取的节点

private static void selfInterrupt() ：自我中断

private final boolean parkAndCheckInterrupt() ： park 并判断线程是否中断

从源码可以看出AQS实现基本的功能：

1.同步器基本范式、结构

2.线程的阻塞、唤醒机制

3.线程阻塞队列的维护

AQS虽然实现了acquire，和release方法，但是里面调用的tryAcquire和tryRelease是由子类来定制的。可以认为同步状态的维护、获取、释放动作是由子类实现的功能，而动作成功与否的后续行为时有AQS框架来实现。

ReentrantLock原理

有了AQS基础，下面来看ReentrantLock的基本原理：

ReentrantLock原理

由于同步器里已经定义了基本的结构，包括获取、释放、和阻塞队列维护和管理等。ReentrantLock是一个独占互斥锁，里只需要实现TryAcquire、TryRelease等方法，告诉同步器是否获取和释放状态成功。其他的后续行为都由AQS框架完成。由于ReentrantLock是一个可重入的独占锁，所以同步器状态可以直接根据是否==0来判断是否可用。

ReentrantLock主要提供lock和unlcok两个方法。

而lock和unlock正是基于AQS的一个子类同步器来实现。里面sync同步器有两种实现，一种是公平锁，一种是非公平锁。默认是非公平锁，看看非公平锁实现tryAcquire

           final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {//如果状态位为0，那么尝试获取
                if (compareAndSetState(0, acquires)) {//基于CAS获取和修改状态
                    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;//其他情况下代表获取失败
        }

再看看公平锁的tryAcquire

            protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (isFirst(current) &&
                    compareAndSetState(0, acquires)) {//判断是否是第一个线程，是的话才尝试获取锁
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

可以看看非公平锁的不会根据FIFO，而公平锁会判断是否是第一个线程，根据FIFO来执行。

参考文献：

The java.util.concurrent Synchronizer Framework

Java 理论与实践: 非阻塞算法简介

《深入浅出 Java Concurrency》目录