Java 并发编程之锁的实现

Java 并发学习之锁的实现

锁是一个同步组件，保证多线程下共享变量的操作是同步的。来看看 Java 自带的 ReentrantLock 的实现。

public class ReentrantLock implements Lock, java.io.Serializable {
    //...
    abstract static class Sync extends AbstractQueuedSynchronizer {
    }
}

可以看出 ReentrantLock 实现了 Lock 接口，同时它还有一个静态内部类 Sync，Sync 继承了 AbstractQueuedSynchronizer 接口。要了解一个锁是如何实现的，首先要知道 Lock 接口和 AbstractQueuedSynchronizer 接口都有什么内容。

Lock 接口 API

public interface Lock {
    //当前线程调用此方法获取锁
    void lock();
    //与 lock() 的区别在与此方法可以响应中断
    void lockInterruptibly() throws InterruptedException;
    //非阻塞获取锁，调用后立即返回，如果成功则返回 true，如果失败则返回 false
    boolean tryLock();
    //超时获取锁，可以响应中断。
    //超时时间内成功获得锁立即返回，超时时间到达后立即返回
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    //释放锁
    void unlock();
    //获取等待通知组件并与该锁绑定
    Condition newCondition();
}

AbstractQueuedSynchronizer 类

AQS（队列同步器）封装了对同步队列的操作，继承这个类实现同步器，需要重写它的 protected 方法。

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    	private volatile int state;
    	//...
    	//独占式地获取同步状态
        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();
   		}
   		//判断同步器是否被当前线程独占
        protected boolean isHeldExclusively() {
        	throw new UnsupportedOperationException();
    	}
    	//...
        protected final int getState() {
        	return state;
    	}
        protected final void setState(int newState) {
        	state = newState;
    	}
        protected final boolean compareAndSetState(int expect, int update) {
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    }

没有重写的 protected 方法如果调用会抛出 UnsupportedOperationException，代表这个方法不被此同步器支持。同步状态是一个由 volatile 修饰的 int 类型的值，AQS 提供了三个 protected 作用域修饰的方法用于操作同步变量，其中 compareAndSetState 方法使用 CAS 设置状态，可以保证操作的原子性。

AQS 同步队列实现

AQS 使用同步队列来管理同步状态，它将队列操作封装进了模板方法，这些队列操作是同步器的核心。

同步队列：双向链表，FIFO，每一个链表元素是一个节点。

节点：用于保存获取同步状态失败的线程引用、等待状态以及前驱和后继节点。

static final class Node {
    static final Node SHARED = new Node();
    static final Node EXCLUSIVE = null;
    //等待状态
    volatile int waitStatus;
    //前驱节点
    volatile Node prev;
	//后继节点
    volatile Node next;
	//获取同步状态失败的线程
    volatile Thread thread;

    Node nextWaiter;

    final boolean isShared() {
        return nextWaiter == SHARED;
    }
	//获取前驱节点
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {    // Used to establish initial head or SHARED marker
    }

    Node(Thread thread, Node mode) {     // Used by addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
    }

    Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

AQS 中包含链表中首尾节点的引用，所有未能获取同步状态的线程被封装成节点后以线程安全的方式插入链表尾部，AQS 中使用 compareAndSetTail 方法来确保插入操作是线程安全的。头节点是成功获取同步状态的节点，头节点在释放同步状态的时候会唤醒后继节点，后继节点在成功获取同步状态的时候会将被设置为头节点，由于只有一个线程能够获取同步状态，因此设置头节点的方法是不需要使用 CAS 保证的。

private transient volatile Node head;
private transient volatile Node tail;

AQS 模板方法

独占式同步状态获取与释放

同步状态获取流程：

如获取成功则退出返回。

如获取失败，则生成节点，将节点加入队列尾部并对该节点调用 acquireQueued 方法。

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)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

自旋过程：acquireQueued 方法首先会判断该节点的前驱节点是否为头节点，如果不是则让线程进入等待状态（可以响应中断或前驱节点的唤醒），如果是，则调用重写的 tryAcquire 方法进行同步状态获取，获取成功后设置头节点为本节点，获取失败后则让线程进入等待状态（可以响应中断或前驱节点的唤醒），当等待的线程被前驱节点唤醒时，会再次进入自旋过程。

同步状态释放：

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

在释放同步状态时，同步器会调用重写的 tryRelease 方法来释放同步状态，然后唤醒头节点的后继节点。

共享式同步状态获取与释放

有一个特殊的节点 Node.SHARED 节点标识该节点在共享模式等待。

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

acquireShared 方法会调用重写的 tryAcquireShared 方法来获取共享式同步状态，如果获取失败，则调用 doAcquireShared 方法进行自旋。

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

该方法会将当前线程以及它的等待状态创建为节点添加到同步队列中，然后执行自旋过程。

自旋过程：首先判断前驱节点是否为头节点，如果是则调用 tryAcquireShared 方法获取共享式同步状态，如果获取成功则设置该节点为头节点并返回，如果获取失败，则让该线程进入等待状态。如果前驱节点不是头节点，则让该线程进入等待状态。当等待的线程被前驱节点唤醒时，会再次进入自旋过程。

独占式超时获取同步状态

使用 doAcquireNanos 可以超时获取同步状态。如果在指定的时间内获取同步状态成功则返回 true，否则返回 false。（传统 synchronized 不具备超时获取特性）。在 JDK 5 之前，当一个线程获取不到锁，它会被阻塞在 synchronized 之外，且无法响应中断操作。JDK 5 中，同步器提供了 acquireInterruptibly 方法，这个方法允许获取同步状态的线程响应中断。doAcquireNanos 相比 acquireInterruptibly 多了一个能够超时获取的特性。

private boolean doAcquireNanos(int arg, long nanosTimeout)
    throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

与独占式同步状态获取相比，独占式超时同步状态获取区别在于对于获取失败后的操作有所不同。

如果超时时间未到，则会等待剩余超时时间，等待完毕后根据中断标志位响应中断，如果未被中断，则重新计算超时时间，重新进行超时判断；如果超时时间已过则会直接返回，取消同步状态的获取。