JAVA中的锁(一)

public static void main(String[] args) {
	Lock lock = new ReentrantLock();
	lock.lock();
	try {
		// TODO
        // 不要将获取锁的操作写在try块中，因为如果在获取锁(自定义锁的实现)时发生了异常
        // 异常发生的同时，也会导致锁无辜释放
	} finally {
		lock.unlock(); //保证锁最终能够尽量释放
	}
}

//获取锁，调用该方法当前线程会获取锁，获取到之后从该方法返回
void lock()

//可中断地获取锁，和lock方法不同的是该方法会响应中断，即在锁的获取中可以中断当前线程
void lockInterruptibly() throws InterruptedException

//尝试非阻塞的获取锁，调用该方法后立即返回，如果能够获取则返回true，否则false
boolean tryLock()

//超时地获取锁当前线程在以下3中情况下会返回：
//当前线程在超时时间内获取了锁
//当前线程在超时时间内被中断
//超时时间结束，返回false
boolean tryLock(long time, TimeUnit unit) throws InterruptedException

//释放锁
void unlock()

//获取等待通知组件，该组件和当前的锁绑定，当前线程只有获得了锁，才能调用该组件的wait方法，
//而调用之后当前线程则释放锁。
Condition newCondition()

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
public class Mutex implements Lock {
	/**
	 * 静态内部类：自定义同步器
	 */
	private static class Sync extends AbstractQueuedSynchronizer {
		private static final long serialVersionUID = 1L;

		@Override
		protected boolean tryAcquire(int arg) { //当状态为0的时候获取锁
			if (compareAndSetState(0, 1)) {
				setExclusiveOwnerThread(Thread.currentThread());
				return true;
			}
			return false;
		}

		@Override
		protected boolean tryRelease(int arg) { //释放锁并将同步状态设置为0
			if (getState()==0) {
				throw new IllegalMonitorStateException();
			}
			setExclusiveOwnerThread(null);
			setState(0);
			return true;
		}

		@Override
		protected boolean isHeldExclusively() { //是否处于独占状态
			return getState()==1;
		}
	}
	
	//仅需将操作代理到Sync上即可
	private final Sync sync = new Sync();
	
	
	@Override
	public void lock() {
		sync.acquire(1);
	}
	
	public boolean isLocked() {
		return sync.isHeldExclusively();
	}

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

	@Override
	public boolean tryLock() {
		return sync.tryAcquire(1);
	}

	@Override
	public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
		return sync.tryAcquireNanos(1, unit.toNanos(timeout));
	}

	@Override
	public void unlock() {
		sync.release(1);
	}

	@Override
	public Condition newCondition() {
		return sync.new ConditionObject();
	}
	
	public boolean hasQueuedThreads() {
		return sync.hasQueuedThreads();
	}
}

static final class Node {  
        /** 标识一个这个节点是否为shared类型 */  
        static final Node SHARED = new Node();  
        /** Marker to indicate a node is waiting in exclusive mode */  
        static final Node EXCLUSIVE = null;  
  
        /** waitStatus value to indicate thread has cancelled */  
        static final int CANCELLED =  1;  
        /** waitStatus value to indicate successor's thread needs unparking */  
        static final int SIGNAL    = -1;  
        /** waitStatus value to indicate thread is waiting on condition */  
        static final int CONDITION = -2;  
        /** 
         * waitStatus value to indicate the next acquireShared should 
         * unconditionally propagate 
         */  
        static final int PROPAGATE = -3;  
  
        /** 
         *   等待状态值，又以下状态值: 
         * 
         *   SIGNAL:     值为-1 ，后续节点处于等待状态，而当前节点的线程如果 
         *               释放了同步状态或者取消等待，节点进入该状态不会变化           
         *   CANCELLED:  值为 1，由于在同步队列中等待的线程等待超时或者被中断 
         *               需要从同步队列中取消等待，节点进入该状态将不会变化                     
         *   CONDITION:  值为-2，节点在等待队列中，节点线程等待在Condition上， 
         *               当其他线程对Condition调用了signal()方法后，该节点将会 
         *               从等待队里中转移到同步队列中，加入对同步状态的获取中 
         
         *   PROPAGATE:  值为-3，表示下一次共享式同步状态获取将会无条件地被传播下去 
         *             
         *   0:          初始化状态 
         */  
        volatile int waitStatus;  
  
        /** 
         * 前驱节点，当节点加入同步队列时被设置 
         */  
        volatile Node prev;  
  
         /** 
         * 后继节点 
         */  
        volatile Node next;  
  
        /** 
         * 获取状态状态的线程 
         */  
        volatile Thread thread;  
  
         /** 
         * 等待队列中的后继节点。如果当前节点是共享的，那么这个字段是一个shared常量， 
         * 也就是说节点类型（独占或共享）和等待队列中个后继节点共用同一个字段 
         */  
        Node nextWaiter;  
  
        /** 
         * Returns true if node is waiting in shared mode. 
         */  
        final boolean isShared() {  
            return nextWaiter == SHARED;  
        }  
  
        /** 
         * Returns previous node, or throws NullPointerException if null. 
         * Use when predecessor cannot be null.  The null check could 
         * be elided, but is present to help the VM. 
         * 
         * @return the predecessor of this node 
         */  
        final Node predecessor() throws NullPointerException {  
            Node p = prev;  
            if (p == null)  
                throw new NullPointerException();  
            else  
                return p;  
        }  
  
        Node() {    // Used to establish initial head or SHARED marker  
        }  
  
        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;  
        }  
    } 

//主要完成同步状态的获取、节点的构造、加入同步队列、以及在同步队列中进行自旋等待的相关工作
//其主要逻辑是：首先调用自定义同步器实现 tryAcquire(int arg) 的方法，该方法保证线程安全
//地获取同步状态，如果同步状态获取失败，则构造同步节点(独占式Node.EXCLUSIVE,同一时刻只能
//有一个线程成功获取同步状态)并通过 addWaiter(Node node)方法将该节点加入到同步队列的尾部，
//最后调用 acquireQueued(Node node, int arg) 方法，使得该节点以“死循环“的方式获取同步
//状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒需要依靠前驱节点的出队 或者阻塞线程
//的中断来实现
public final void acquire(int arg) {
  	if (!tryAcquire(arg) &&
     	 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    	selfInterrupt();
}


//compareAndSetTail(Node expect, Node update)保证了节点能够被线程安全的添加。试想一下，
//如果使用一个普通的LinkedList来维护节点之间的关系，那么当一个线程获取了同步状态，而其他线程
//由于调用 tryAcquire(int arg) 方法获取同步状态失败而被并发地添加到Linkedlist中时，
//LinkedList将难以保证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;
    	}
  	}
  	enq(node);
  	return node;
}

//enq(final Node node)方法中，同步器通过”死循环”来保证节点的正确添加，在“死循环”中只有
//通过CAS将节点设置为尾节点之后，当前线程才能从该方法中返回，否则当前线程不断地尝试设置。可以
//看出，该方法将并发添加节点的请求通过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;
            }
        }
    }
}

//节点进入到同步队列之后，就进入了一个自旋的过程，每个节点（或者说每个线程）都在自省地观察，
//当条件满足获取到同步状态时，就可以从这个自旋的过程中退出，否则依旧留在这个自旋的过程中（
//并且阻塞节点的线程），代码如下。在该方法中当前线程在“死循环”中尝试获取同步状态，而只有
//前驱节点是头结点才能尝试获取同步状态（这样保证了早获取早通知的逻辑思路）。
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;
            }	
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

//该方法释放了同步状态之后，会唤醒其后继节点(进而使得后继节点能够重新尝试获取同步状态)
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg); //获取同步状态失败后
}

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg); //如果当前节点的前驱节点是头结点，尝试获取同步状态
                if (r >= 0) {				   //获取同步状态成功后
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

//该方法在释放同步状态之后，将会唤醒后续处于等待状态的节点
//对于能够支持多个线程同时访问的并发组件(e.g. Semaphore)，在释放资源上它和独占式的主要区别在于
//tryReleaseShared(int arg)方法必须确保同步状态(或者资源数)线程被安全释放，一般是通过循环和CAS
//来保证的，因为释放同步状态的操作会来自于多个线程。
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;
    }
}

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class TwinsLock implements Lock {
	private final Sync sync = new Sync(2);
	
	private static final class Sync extends AbstractQueuedSynchronizer {
		private static final long serialVersionUID = 1L;

		Sync(int count) {
			if (count < 0) {
				throw new IllegalArgumentException("count must large than 0");
			}
			setState(count);
		}

		//返回值大于0才表示当前线程才获取到了同步状态
		@Override 
		protected int tryAcquireShared(int reduceCount) {
			for (;;) {
				int current = getState();
				int newCount = current-reduceCount;
				if (newCount<0 || compareAndSetState(current, newCount)) { //CAS确保正确设置
					return newCount;
				}
			}
		}

		@Override
		protected boolean tryReleaseShared(int returnCount) {
			for(;;) {
				int current = getState();
				int newCount = current + returnCount;
				if (compareAndSetState(current, newCount)) { //CAS确保正确设置
					return true;
				}
			}
		}
	}
	
	/**
	 * 获取锁
	 */
	@Override
	public void lock() {
		sync.acquireShared(1);
	}

	/**
	 * 释放锁
	 */
	@Override
	public void unlock() {
		sync.releaseShared(1);
	}
	
	@Override
	public void lockInterruptibly() throws InterruptedException {
		// TODO
	}

	@Override
	public boolean tryLock() {
		// TODO
		return false;
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		// TODO
		return false;
	}

	@Override
	public Condition newCondition() {
		// TODO 
		return null;
	}
}

public class TwinsLockTest {
	public static void main(String[] args) throws InterruptedException {
		final Lock lock = new TwinsLock();
		class Worker implements Runnable {
			@Override
			public void run() {
				lock.lock();
				try {
					sleep(1);
					System.out.println(Thread.currentThread().getName());
					sleep(1);
				} finally {
					lock.unlock();
				}
			}
		}
		
		//启动10个线程
		for (int i = 0; i < 10; i++) {
			Worker worker = new Worker();
			Thread thread = new Thread(worker, "Thread-"+i);
			thread.setDaemon(true);
			thread.start();
		}
		sleep(10);
	}
	
	public static void sleep(long seconds) {
		try {
			Thread.sleep(seconds*1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

//运行结果
//运行测试可以看到线程名称成对输出，也就是在同一个时刻只有两个线程能够获取到锁，
//这表明TwinsLock可以按照预期正常工作。