Lock锁之公平锁与非公平锁(AQS实现原理)

锁的可重入性

在Concurrent包中的锁都是可重入锁，一般都命名为ReentrantX。可重入锁是指当一个线程调用object.lock拿到锁，进入互斥区后，再次调用object.lock，仍然可以拿到该锁。
synchtonized关键字就是可重入锁。

ReentrantLock的类继承结构

Lock中定义的模版方法

public interface Lock {
    void lock();
    void lockInterruptibly() throws InterruptedException;
    boolean tryLock();
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    void unlock();
    Condition newCondition();
}
1
2
3
4
5
6
7
8

ReentrantLock中的方法

ReentrantLock本身没有代码逻辑，实现都在Sync内部类中

public class ReentrantLock implements Lock, java.io.Serializable {

    public ReentrantLock() {
        sync = new NonfairSync();
    }

    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
    
    public void lock() {
        sync.lock();
    }

    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    public boolean tryLock() {
        return sync.nonfairTryAcquire(1);
    }
   
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

公平锁与非公平锁

Sync是一个抽象类，它有两个子类FairSync与NonfaitSync，分别对应公平锁和非公平锁，在ReentrantLock的构造函数中可以看出来：

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}
1
2
3

对于公平以及非公平的概念，举个例子，一个人去做核酸排队，自己排到队伍末尾，这叫公平，直接去插队这叫做非公平，线程也是一样的，直接去抢锁，显然是不公平的。

锁的基本实现原理

要实现一把具有阻塞或唤醒功能的锁，需要以下几个核心要素：

1. 需要一个state变量，标记该锁的状态，state变量至少有两个值，0、1。对state便利的操作要确保线程安全，也就是会用到CAS操作。
2. 需要记录是那个线程持有锁。
3. 需要底层支持对一个线程进行阻塞或者唤醒操作。
4. 需要一个队列维护所有的阻塞线程。这个队列必须是线程安全的无锁队列，也需要用到CAS。
   对于1，2在类AbstractOwnableSynchronizer以及AbstractQueuedSynchronizer中有体现：

public abstract class AbstractOwnableSynchronizer
    implements java.io.Serializable {

      //记录锁被哪个线程占用
    private transient Thread exclusiveOwnerThread;
}
public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    //记录锁的状态，通过CAS来改变
    private volatile int state;
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13

state的取值不仅可以是0，1；还可以大于1，就是为了支持锁的可重入性：

• 当state=0，没有线程持有锁，exclusiveOwnerThread为null
• 当state=1，有一个线程持有锁，exclusiveOwnerThread为该线程
• 当state>1, 说明该线程重入了该锁

对于3，在Unsafe中提供了阻塞唤醒的原语操作：

public class LockSupport {
    private LockSupport() {} 

    public static void unpark(Thread thread) {
        if (thread != null)
            UNSAFE.unpark(thread);
    }

    public static void park() {
        UNSAFE.park(false, 0L);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13

当在线程中调用Park，改线程就会被阻塞，在另外一个线程中调用unpark，就可以唤醒阻塞在park地方的线程
对于4，在AQS中利用双向链表和CAS来实现一个阻塞队列：

static final class Node {
 
    volatile Node prev;
    volatile Node next;
    volatile Thread thread;
    Node nextWaiter;
    
    private transient volatile Node head;
    private transient volatile Node tail;

}
1
2
3
4
5
6
7
8
9
10
11

阻塞队列是整个AQS的核心，如下图所示：head指向双向链表头部，tail指向双向链表的尾部，入队就是将信的node放入到tail后面，然后对tail进行CAS操作，出队就是对head进行CAS操作，把head向后移动一个位置

公平锁lock的实现

直接查看ReentrantLock的lock方法，发现它其实调用的时Sync的lock方法，公平锁的lock方法实现在内部类FairSync中，FairSync中的lock方法调用AbstractQueuedSynchronizer类中的acquire方法：

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
1
2
3
4
5

• 先分析一下tryAcquire方法，这个方法的逻辑实际在FairSync类中：

protected final boolean tryAcquire(int acquires) {
        // 获取当前线程
        final Thread current = Thread.currentThread();
        // 获取锁标志
        int c = getState();
        // 0标识锁没有被占用，尝试获取锁
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                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;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在公平锁的tryAcquire方法比非公平锁多了一个逻辑，就是hasQueuedPredecessors方法，这个方法的意思是，在锁未被占用时，通过hasQueuedPredecessors方法判断当前线程如果排在队列的第一个(队列无其他线程)，才会尝试去抢锁，否则继续排队。

• tryAcquire尝试获取锁，如果没有获取到锁，则会返回false，调用acquireQueued(addWaiter(Node.EXCLUSIVE))的方法，先看addWaiter(Node.EXCLUSIVE)方法

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        //先尝试加到队列尾部，如果不成功，执行下面的enq
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}
//进行队列的初始化，新建一个空的node，然后不断尝试自旋，直到成功将该节点加入到队列尾部为止
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // 初始化
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            // 自旋将节点加入到队列尾部
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
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

addWaiter将线程加入到队列尾部之后，执行acquireQueued方法，线程一进入到该方法就会被阻塞，即使有其他线程调用了interrpt方法也不能将其唤醒，除非其他线程释放了锁，并且该线程拿到了锁，才会从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;
            }
            //调用park中断自己，并且判断是否收到中断信号
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

在该节点加入到阻塞队列之后，判断它前一个节点是不是head，如果是，说明该节点是队列中的唯一一个节点则尝试获取锁，否则调用park中断自己，进入阻塞，等待其它线程调用unpark释放锁后继续执行。

公平锁的unlock实现

ReentrantLock的unLock方法和lock方法一样也没有逻辑，都是调用Sync的父类AbstractQueuedSynchronizer中的release方法，最后调用到Sync的tryRelease

AbstractQueuedSynchronizer#release

public final boolean release(int arg) {
    //调用tryRelease释放锁成功后
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            //调用unparkSuccessor释放锁
            unparkSuccessor(h);
        return true;
    }
    return false;
}
1
2
3
4
5
6
7
8
9
10
11

Sync#tryRelease释放锁

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    // 只有锁的拥有者才可以释放锁
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}
1
2
3
4
5
6
7
8
9
10
11
12
13

非公平锁lock的实现

非公平锁相比校公平锁的获取就是调用lock方法时就尝试获取锁，这就非公平锁的体现的不公平性

final void lock() {
    // 尝试获取锁
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}
1
2
3
4
5
6
7