> 科技资讯 > AQS源码学习

AQS源码学习

科技资讯 投稿 6500℃ 0 评论

AQS源码学习

抽象队列同步器AQS

AQS介绍

FIFO队列 维护线程的同步状态，实现类只需要继承 AbstractQueuedSynchronizer，并重写指定方法(tryAcquire(/tryRelease(等即可实现线程同步机制。

AQS 继承结构

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable{
    //...
}

AbstractOwnableSynchronizer

独占资源与占有线程的关联

package java.util.concurrent.locks;

public abstract class AbstractOwnableSynchronizer implements java.io.Serializable {
    private static final long serialVersionUID = 3737899427754241961L;
    protected AbstractOwnableSynchronizer( { }
    
    // 占有线程
    private transient Thread exclusiveOwnerThread;
    
    // 设置占有线程
    protected final void setExclusiveOwnerThread(Thread thread {
        exclusiveOwnerThread = thread;
    }
    
    // 获取占有线程
    protected final Thread getExclusiveOwnerThread( {
        return exclusiveOwnerThread;
    }
}

AQS原理

AQS维护一个CLH (Craig, Landin, and Hagersten 双向队列，记录头指针head(头指针无意义，没有对应线程 和 尾指针tail，同时维护了一个 volatile int state 变量记录同步状态(初始状态默认为0，表示未被该资源未被占用。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable{
    private static final long serialVersionUID = 7373984972572414691L;
    
    // 队列头节点
    private transient volatile Node head;
    
    // 队列尾节点
    private transient volatile Node tail;
    
    // 同步状态
    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;
    }
}

申请锁 -> lock( 执行过程

以 ReentrantLock 为例子，该锁默认实现是一个非公平独占锁。

public static void main(String[] args {
    // 独占锁、默认非公平锁
	ReentrantLock reentrantLock = new ReentrantLock(;
    reentrantLock.lock(;
}


// ReentrantLock.lock(
public void lock( {
    sync.lock(;
}

// NofairSync.lock(
final void lock( {
    // CAS 获取锁
    if (compareAndSetState(0, 1
        setExclusiveOwnerThread(Thread.currentThread(;
    else
        // 获取锁失败
        acquire(1;
}


public final void acquire(int arg {
    // tryAcquire: 尝试获取锁
    // acquireQueued: 添加到阻塞队列
    if (!tryAcquire(arg &&
        acquireQueued(addWaiter(Node.EXCLUSIVE, arg
        selfInterrupt(;
}

tryAcquire 执行链: 尝试获取锁，获取不到则返回 false


// NofairSync.tryAcquire(
protected final boolean tryAcquire(int acquires {
    return nonfairTryAcquire(acquires;
}


// Sync.nofairTryAcquire(
final boolean nonfairTryAcquire(int acquires {
    final Thread current = Thread.currentThread(;
    int c = getState(;
    if (c == 0 {
        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;
}

acquireQueued 执行链:


    首先通过 addWaiter 方法将线程添加到队列尾部

  1. 然后通过 acquireQueued 方法实现线程进入CLH队列后如何被阻塞或者被唤醒获取锁
    2. 



// AbstractQueuedSynchronizer.addWaiter(
// 添加 node 到等待队列尾部
// 返回插入的节点 node
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;
        if (compareAndSetTail(pred, node {
            pred.next = node;
            return node;
        }
    }
    
    // tail == null
    enq(node;
    return node;
}


// 线程进入等待队列之后，如何获取锁或者继续阻塞
final boolean acquireQueued(final Node node, int arg {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;; {
            final Node p = node.predecessor(;
            
            // 如果当前节点的前驱节点为 head，则竞争锁资源
            if (p == head && tryAcquire(arg {
                setHead(node;
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            
            // 当前节点的前驱节点不是 head, 或者竞争锁失败
            // shouldParkAfterFailedAcquire: true, 调用 parkAndCheckInterrupt( 阻塞线程
            // shouldParkAfterFailedAcquire: false, 再次进入循环块，竞争锁
            if (shouldParkAfterFailedAcquire(p, node &&
                parkAndCheckInterrupt(
                interrupted = true;
        }
    } finally {
        if (failed
            // for循环意外退出才能走到这
            cancelAcquire(node;
    }
}


/**
 * 判断当前线程是否需要阻塞
 * 阻塞(return true:
 *      1.前驱节点的状态 pred.waitStatus=SIGNAL
 * 不阻塞(return false:
 *      1.前驱节点的状态为 CANCELLED，循环向前找 ws <= 0 的前驱节点
 *      2.前驱节点的状态 ws = 0 || ws = PROPAGATE
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0 {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0;
        pred.next = node;
    } else {
        /*
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL;
    }
    return false;
}

释放锁 -> unlock( 执行过程

同样以 ReentrantLock 为例子，该锁默认实现是一个非公平独占锁。

public static void main(String[] args {
    // 独占锁、默认非公平锁
	ReentrantLock reentrantLock = new ReentrantLock(;
    reentrantLock.lock(;
    
    try {
        // 业务代码
    } catch (Exception e {
        e.printStackTrace(;
    } finally {
        reentrantLock.unlock(;
    }
}


// ReentrantLock.unlock(
public void unlock( {
    sync.release(1;
}

release 执行链:


    通过 tryRelease 方法判断当前锁是否已经被完全释放

  1. 如果已经被完全释放 -> 则唤醒其后继节点对应的线程
    2. 



// AbstractQueuedSynchronizer.release(
// tryRelease( 返回 true -> 则执行 if 中的逻辑 -> unparkSuccessor: 唤醒后继节点
public final boolean release(int arg {
    if (tryRelease(arg {
        Node h = head;
        if (h != null && h.waitStatus != 0
            unparkSuccessor(h;
        return true;
    }
    return false;
}


// 释放锁，修改 state
// free: true 锁已经完全释放，唤醒其他线程竞争
// free: false 锁仍然被当前线程占有
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;
}

unparkSuccessor: 唤醒后继节点


    情况1：直接唤醒当前节点的后继节点

  • 情况2: 情况1对应的节点状态为 CANCELLED，则从CLH队列尾部开始寻找 ws <= 0 的节点唤醒
    • 



/**
 * 唤醒后继节点
 * 
 * waitStatus:
 *     CANCELLED(1  : 当前节点因超时或响应中断结束调度，进入该状态的节点不再变化
 *     SIGNAL(-1    : 后继节点等待当前节点唤醒
 *     CONDITION(-2 : 当前节点处于 condition 上，等待转移到CLH同步队列
 *     PROPAGETE(-3 : 当前节点处于 shared 状态
 *     0             : 默认状态
 */
private void unparkSuccessor(Node node {
    /*
     * If status is negative (i.e., possibly needing signal try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0
        compareAndSetWaitStatus(node, ws, 0;

    // 情况1：直接唤醒当前节点的后继节点
    // 情况2: 情况1对应的节点状态为 CANCELLED，则从CLH队列尾部开始寻找 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;
}

FairSync && NofairSync

FairSync: 以 ReentrantLock 的公平锁实现为例


static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock( {
        acquire(1;
    }

    protected final boolean tryAcquire(int acquires {
        final Thread current = Thread.currentThread(;
        int c = getState(;
        if (c == 0 {
            // 判断 CLH 队列是否有正在等待的线程，如果有，则唤醒CLH 队列 head 的后继节点
            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;
    }
}

NofairSync: 以 ReentrantLock 的非公平锁实现为例


static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock( {
        // CAS 抢锁，如果恰巧没有线程占有，则直接获取锁返回
        if (compareAndSetState(0, 1
            setExclusiveOwnerThread(Thread.currentThread(;
        else
            // 抢锁失败，则进入 acquire
            acquire(1;
    }

    protected final boolean tryAcquire(int acquires {
        return nonfairTryAcquire(acquires;
    }
}


final boolean nonfairTryAcquire(int acquires {
    final Thread current = Thread.currentThread(;
    int c = getState(;
    if (c == 0 {
        // 同样进行 CAS 抢锁，而不是判断 CLH 队列中是否有等待线程
        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;
}

所以，公平锁和非公平锁的区别总结如下：

    非公平锁调用 lock( 方法，会马上进行一次 CAS 抢占锁
  1. 抢占锁失败后进入 tryAcquire( 方法，公平锁会去判断CLH等待队列是否有线程处于等待状态，如果有则不抢占锁；非公平锁则会直接进行 CAS 尝试抢占锁

[^]: 注：以上源代码阅读与分析，基于 Oracle JDK8 版本

编程笔记 » AQS源码学习

赞同 (27) or 分享 (0)
游客 发表我的评论   换个身份
取消评论

表情
(0)个小伙伴在吐槽