多线程(八) AQS

// 前驱节点
private transient volatile Node head;

// 后驱节点
private transient volatile Node tail;

// 静态内部类Node
static final class Node {

	/** 共享锁 */
	static final Node SHARED = new Node();

	/** 独占锁 */ 
	static final Node EXCLUSIVE = null;

	/** 表示线程已被取消 */
	static final int CANCELLED = 1;

	/** 表示后续线程需要取消阻塞 */
	static final int SIGNAL = -1;

	/** 表示线程在条件下等待 */
	static final int CONDITION = -2;

	/** 表示下一个获取共享应无条件传播 */
	static final int PROPAGATE = -3;

	/** 
	 * 节点等待状态
	 * 等于0:该节点尚未被初始化完成
	 * 大于0:说明该线程中断或者等待超时，需要移除该线程
	 * 小于0:该线程处于可以被唤醒的状态
	 */
	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() {
	}

       // 构造器
	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;
	}
}

private Node addWaiter(Node mode) {
	// 新建Node
	Node node = new Node(Thread.currentThread(), mode);
	// CAS快速尝试添加尾节点(侥幸心理，万一成功了呢)
	Node pred = tail;
	if (pred != null) {
			node.prev = pred;
		//CAS设置尾节点
		if (compareAndSetTail(pred, node)) {
			pred.next = node;
			return node;
		}
	}
	//多次尝试
	enq(node);
	return node;
} 

private Node enq(final Node node) {
       //多次尝试，直到成功为止
       for (;;) {
           Node t = tail;
           //tail不存在，设置为首节点
           if (t == null) {
               if (compareAndSetHead(new Node()))
                   tail = head;
           } else {
               //设置为尾节点
               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 (;;) {

			//拿到前驱节点
			final Node p = node.predecessor();

			/* 
			 * 如果前驱是head，说明自己排在第二位，有可能马上就被执行
           	 * 所以再次尝试tryAcquire()获取，如果失败就挂起等待
           	 * 当然有可能是第一位搞完了释放资源唤醒自己，也有可能被interrupt
           	 */
			if (p == head && tryAcquire(arg)) {

				// 获取到资源后，把自己设置为head，也就是说head指向的永远是当前拿到资源的
				setHead(node);

				// 断绝与前驱节点的联系，方便被GC回收
				p.next = null;

				// 成功获取资源后将失败标记为false
				failed = false;

				// 返回等待过程中是否被中断过
				return interrupted;
			}

			/*
			 * 先去检查自己是否真的可以被挂起了，如果不符合条件会进入下一次循环直到符合为止
			 * 调用park()方法将自己挂起，直到被唤醒
			 * 唤醒后会返回是否被中断标记，方便下次return出去
			 */
			if (shouldParkAfterFailedAcquire(p, node) &&
				parkAndCheckInterrupt())
				interrupted = true;
		}
	} finally {

		// 如果等待过程中没有成功获取资源(如timeout，或者可中断的情况下被中断了)，取消结点在队列中的等待。
		if (failed)
			cancelAcquire(node);
	}
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

	// 拿到前驱节点的状态
	int ws = pred.waitStatus;

	// 如果前驱节点的状态是SIGNAL，那么前驱节点执行完会自动唤醒自己，放心的将自身挂起就好了
       if (ws == Node.SIGNAL)
           return true;

       // 如果前驱节点执行过程中放弃了(超时或者其他的)，一直往前找，直到找到正常等待的状态节点
       if (ws > 0) {
           do {
               node.prev = pred = pred.prev;
           } while (pred.waitStatus > 0);
           pred.next = node;
       } else {
           // 如果前驱正常，那就把前驱的状态设置成SIGNAL，告诉它拿完号后通知自己一下
           compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
       }
       return false;
   }

private final boolean parkAndCheckInterrupt() {

	//调用park()使线程进入waiting状态
	LockSupport.park(this);

	//如果执行到这里，说明被唤醒，查看自己是不是被中断的。
	return Thread.interrupted();
}

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

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) {

	// 获取当前节点的状态
       int ws = node.waitStatus;

       // 如果小于0，使用CAS设置为0，0代表退出等待队列
       if (ws < 0)
           compareAndSetWaitStatus(node, ws, 0);

       // 获取后继节点
       Node s = node.next;

       // 如果没有后继节点，或者后继节点状态大于0，也就是说已经退出队列了
       if (s == null || s.waitStatus > 0) {

       	// 方便GC回收
           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);
   }

public final void acquireInterruptibly(int arg)
           throws InterruptedException {
       if (Thread.interrupted())
           throw new InterruptedException();
       if (!tryAcquire(arg))
           doAcquireInterruptibly(arg);
   }

private void doAcquireInterruptibly(int arg)
       throws InterruptedException {
       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;
               }
               if (shouldParkAfterFailedAcquire(p, node) &&
                   parkAndCheckInterrupt())
                   throw new InterruptedException();
           }
       } finally {
           if (failed)
               cancelAcquire(node);
       }
   }

   private final boolean parkAndCheckInterrupt() {
       LockSupport.park(this);
       return Thread.interrupted();
   }

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

public final boolean releaseShared(int arg) {
       if (tryReleaseShared(arg)) {
           doReleaseShared();
           return true;
       }
       return false;
   }

private void doReleaseShared() {
	for (;;) {
           Node h = head;
           if (h != null && h != tail) {
               int ws = h.waitStatus;
               if (ws == Node.SIGNAL) {
                   if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                       continue;            // loop to recheck cases
                   unparkSuccessor(h);
               }
               else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                   continue;                // loop on failed CAS
           }
           if (h == head)                   // loop if head changed
               break;
       }
}

public class CustomLock {

	private Sync sync;

	// 自定义同步器
	private static class Sync extends AbstractQueuedSynchronizer {

		// 判断是否锁定状态
		@Override
		protected boolean isHeldExclusively() {
			return getState() == 1;
		}

		// 获取资源
		@Override
		protected boolean tryAcquire(int arg) {

			// 使用CAS修改状态，如果成功设置当前资源为独占资源
			if(compareAndSetState(0, 1)){
				setExclusiveOwnerThread(Thread.currentThread());
				return true;
			}
			return false;
		}

		// 释放资源
		@Override
		protected boolean tryRelease(int arg) {

		//既然释放，肯定就是已占有状态了，为了代码健壮一点加层判断
		if (getState() == 0)
			throw new IllegalMonitorStateException();

		// 清空独占记录
		setExclusiveOwnerThread(null);

		// 释放共享资源，tryRelease还没执行完，线程仍然持有锁，因此不需要CAS修改
		setState(0);

		return true;
		}
	}

	// 在自定义加锁对象创建时，为其初始化一个同步器
	public CustomLock(){
		sync = new Sync();
	}

	// 加锁
	public void lock() {
		sync.acquire(1);
	}

	// 单次加锁尝试
	public boolean tryLock() {
		return sync.tryAcquire(1);
	}

	// 释放锁
	public void unlock(){
		sync.release(1);
	}

	// 锁是否处于加锁状态
	public boolean isLocked(){
		return sync.isHeldExclusively();
	}
}