简介
-
AbstractQueuedSynchronizer是多线程编程中程序同步的核心，所以值得深究原理。以下以ReentrantLock中NonfairSync演示程序执行流程。IDEA中默认不能debug到JDK的源码。可用以下设置。
idea设置debug不跳过包 cas算法：compare and set，在指定内存地址中更新值时，需要对比该地址中的当前值是和期望值一致，才能更新成功。-
NonfairSync类图
NonfairSync类图
加锁流程
AbstractQueuedSynchronizer内部维护一个链表。head是获取锁的节点，第二节点是等待获取锁，然后想要获取锁就插入链表尾部等待。
addWaiter方法循环不停的在链表尾部插入等待者。
acquireQueued方法循环不停的去除链表首部，如果首部释放锁，立即将首部去除，第二节点获取锁，成功成为首部。如果没有释放锁，则各等待者进入wait状态。
package com.redoor.lock;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
public static void main(String[] args) {
ReentrantLock lock = new ReentrantLock();
for (int i = 0; i < 3; i++) {
new Thread(new SubThread(lock, i)).start();
}
}
}
package com.redoor.lock;
import java.util.concurrent.locks.ReentrantLock;
public class SubThread implements Runnable {
private ReentrantLock lock;
private int i;
public SubThread(ReentrantLock lock, int i) {
this.i = i;
this.lock = lock;
}
@Override
public void run() {
try {
lock.lock();
i++;
System.out.println(Thread.currentThread().getName() + ":" + i);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
lock.lock()调用NonfairSync中的lock()
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
//cas失败
acquire(1);
}
cas设置AbstractQueuedSynchronizer中statue为1，标记为加锁状态。并设置加锁的线程为当前线程。cas失败后调用AbstractQueuedSynchronize中acquire()
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire会调用nonfairTryAcquire 方法，如果加锁就返回false。重点看addWaiter和acquireQueued
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;
//如果链表tail已经初始化，则直接将等待者插入tail
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
//初始化链表head、tail
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
//循环等待者插入链表tail
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取node的head
final Node p = node.predecessor();
//如果node为等待者，链表中为第二位置，则试着获取锁
//如果获取锁，则将链表中head去除，自己为head
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//获取锁失败，设置waitstate=-1，并将当前线程处于wait状态
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
//当前线程被wake后，继续循环，只到获取锁
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
释放流程
先将锁状态去除，然后唤醒第二节点争取锁。
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;
}
public final boolean release(int arg) {
//去除锁状态
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
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);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
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)
//唤醒第二节点，然后第二节点去除head
LockSupport.unpark(s.thread);
}
自定义锁
AbstractQueuedSynchronizer中定义获取锁和释放锁的模版方法。acquire方法会调用tryAcquire方法，而类中该方法为空方法，方便子类重写获取锁和释放锁的条件。release方法和tryRelease方法也是一样。而对于链表的维护则不用操心。上文中的NonfairSync就是重写了这两个方法。
自定义锁，是实现Lock接口的实现类。类中定义AbstractQueuedSynchronizer的子类，并重写tryAcquire和tryRelease方法。
private class Sync extends AbstractQueuedSynchronizer{
@Override
protected boolean tryAcquire(int arg) {
Thread thread = Thread.currentThread();
int state = getState();
if (state == 0) {
compareAndSetState(0, arg);
setExclusiveOwnerThread(thread);
return true;
} else if (getExclusiveOwnerThread() == thread) {
compareAndSetState(state, state + arg);
return true;
}
return false;
}
@Override
protected boolean tryRelease(int arg) {
Thread thread = Thread.currentThread();
if (getExclusiveOwnerThread() != thread) {
throw new RuntimeException();
}
boolean flag = false;
int state = getState() - 1;
if (state == 0) {
flag = true;
setExclusiveOwnerThread(null);
}
setState(state);
return flag;
}
}
public class LucyLock implements Lock {
private Sync sync = new Sync();
@Override
public void lock() {
sync.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return sync.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
sync.release(1);
}
}
测试类
public class Main {
private int value;
private LucyLock lock = new LucyLock();
private void next(){
try {
lock.lock();
Thread.sleep(300);
value++;
//up();//测试重入锁
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
private void up() {
try {
lock.lock();
System.out.println("up");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public static void main(String[] args) {
Main main = new Main();
for (int i = 0; i < 100; i++) {
new Thread(()->{
while (true) {
main.next();
System.out.println(Thread.currentThread().getName() + ":" + main.value);
}
}).start();
}
}
}
