AQS解读及其实践

// 同步队列的head节点, 延迟初始化，除了初始化，只能通过setHead方法修改
// 如果head存在，waitStatus一定是CANCELLED
private transient volatile Node head;
// 同步队列的tail节点，延迟初始化，只能通过enq方法修改
private transient volatile Node tail;
// 同步状态
private volatile int state;
// 支持CAS
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
private static final long waitStatusOffset;
private static final long nextOffset;

    protected final int getState() {
return state;
}protected final void setState(int newState) {
state = newState;
}protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}// 钩子方法，独占式获取同步状态, 需要子类实现，实现此方法需要查询当前同步状态并
// 判断同步状态是否符合预期，然后再CAS设置同步状态
// 返回值true代表获取成功，false代表获取失败
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}// 钩子方法，独占式释放同步状态，需要子类实现，
// 等待获取同步状态的线程将有机会获取同步状态
// 返回值true代表获取成功，false代表获取失败
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}// 钩子方法，共享式获取同步状态，需要子类实现，
// 返回值负数代表获取失败、0代表获取成功但没有剩余资源、
// 正数代表获取成功，还有剩余资源
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}// 钩子方法，共享式释放同步状态，需要子类实现
// 返回值负数代表获取失败、0代表获取成功但没有剩余资源、
// 正数代表获取成功，还有剩余资源
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}// 模板方法，独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，
// 否则会进入同步队列等待，此方法会调用子类重写的tryAcquire方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}// 模板方法，独占式的释放同步状态，该方法会在释放同步状态后，
// 将同步队列中的第一个节点包含的线程唤醒
// 此方法会调用子类重写的tryRelease方法
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}// 模板方法，共享式的获取同步状态，如果当前系统未获取到同步状态，
// 将会进入同步队列等待，同一时刻可以有多个线程获取到同步状态
// 此方法会调用子类重写的tryAcquireShared方法
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}// 模板方法，共享式的释放同步状态
// 此方法会调用子类重写的tryReleaseShared方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}// 用于将当前线程加入到等待队列的队尾，并返回当前线程所在的结点
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;    // 尝试将Node放到队尾
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}//初始化或自旋CAS直到入队成功
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;
}
}
}
}

public class MyCountDownLatch {private Sync sync;public MyCountDownLatch(int count) {
sync = new Sync(count);
}public void countDown() {
sync.releaseShared(1);
}public void await() {
sync.acquireShared(1);
}class Sync extends AbstractQueuedSynchronizer {
public Sync(int count) {
setState(count);
}@Override
protected int tryAcquireShared(int arg) {
// 只有当state变为0时，加锁成功
return getState() == 0 ? 1 : -1;
}@Override
protected boolean tryReleaseShared(int arg) {
for (; ; ) {
int c = getState();
if (c == 0) return false;
int nextc = c - 1;
// 用CAS操作，讲count减一
if (compareAndSetState(c, nextc)) {
// 当state=0时，释放锁成功，返回true
return nextc == 0;
}
}
}
}
}// 测试
public class MyCountDownLatchTest {
/*
每隔1s开启一个线程，共开启6个线程
若希望6个线程 同时 执行某一操作
可以用CountDownLatch实现
*/
public static void test01() throws InterruptedException {
MyCountDownLatch ctl = new MyCountDownLatch(6);for (int i = 0; i < 6; i++) {
new Thread() {
@Override
public void run() {
ctl.countDown();
ctl.await();
System.out.println("here I am...");
}
}.start();
Thread.sleep(1000L);
}
}/*
开启6个线程，main线程希望6个线程都执行完某个操作后，才执行某个操作
可以用CountDownLatch来实现
*/
public static void test02() throws InterruptedException {
MyCountDownLatch ctl = new MyCountDownLatch(6);for (int i = 0; i < 6; i++) {
new Thread() {
@Override
public void run() {
System.out.println("after print...");
ctl.countDown();
}
}.start();
Thread.sleep(1000L);
}ctl.await();
System.out.println("main thread do something ...");
}public static void main(String args[]) throws InterruptedException {
test01();
}
}

public class MySemaphore {private Sync sync;public MySemaphore(int permits) {
sync = new Sync(permits);
}//抢信号量、就是在加锁
public void acquire() {
sync.acquireShared(1);
}//释放信号量，就是解锁
public void release() {
sync.releaseShared(1);
}class Sync extends AbstractQueuedSynchronizer {
private int permits;public Sync(int permits) {
this.permits = permits;
}@Override
protected int tryAcquireShared(int arg) {
int state = getState();
int nextState = state + arg;
// 如果信号量没占满，加锁的个数没有达到permits
if (nextState <= permits) {
if (compareAndSetState(state, nextState)) {
return 1;
}
}
return -1;
}@Override
protected boolean tryReleaseShared(int arg) {
int state = getState();
if (compareAndSetState(state, state - arg)) {
return true;
} else {
return false;
}
}
}
}// 测试
public class MySemaphoreTest {
static MySemaphore sp = new MySemaphore(6);public static void main(String args[]) {
for (int i = 0; i < 1000; i++) {
new Thread() {
@Override
public void run() {
try {
sp.acquire(); // 抢信号量、就是在加锁
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
queryDB("localhost:3006");
sp.release(); // 释放信号量，就是解锁
}
}.start();
}
}public static void queryDB(String url) {
System.out.println("query " + url);
}
}