聊聊高并发（九）实现几种自旋锁（四）

这篇看一下时限队列锁的一种实现方式。

java并发包中的Lock定义包括了时限锁的接口：

public interface Lock {

    void lock();

    void lockInterruptibly() throws InterruptedException;

    boolean tryLock();

    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

    void unlock();

    Condition newCondition();
}

tryLock就是实现锁的接口，它支持限时操作，支持中断操作。这两个特性非常重要，能够防止死锁，也能够在死锁的情况下取消锁。

由于这两个特性的须要，队列锁的节点须要支持“退出队列”的机制。也就是说当发生超时或者线程中断的情况下。线程能从队列中出队列，不影响其它节点继续等待。之前实现的几种队列锁都不支持退出机制。一旦发生队列中的线程长时间堵塞，那么兴许全部的线程都会被动堵塞。

我们看一种限时队列锁的实现，它有几个要点：

1. 定义一个共享的AVAILABLE节点，当一个节点的preNode指向AVAILABLE时，表示这个节点获得锁

2. QNode节点维护一个preNode引用。这个引用仅仅有当获得锁时。会指向AVAILABLE，或者超时了会指向它的前一个节点，其它等待锁的时候都是Null，由于一旦一个节点超时了。须要让它的兴许节点指向它的前驱节点，所以仅仅有超时的时候会给preNode设置值（指向AVAILABLE节点除外）。

3. 使用一个AtomicReference原子变量tail来形成一个虚拟的单向链表结构。tail的getAndSet操作会返回之前的节点的引用。相当于获得了前驱节点。当获得锁后，前驱节点引用就释放了，前驱节点就能够被GC回收

4. 支持中断操作，Thread.isInterrupted()能够获得线程中断的信息，一旦获取中断信息。就抛出中断异常。须要注意的时，线程中断信息发出时。并非要求线程立即中断。而是告知了线程要中断的信息，程序自己控制中断的地点。

5. 因为线程仅仅有一个ThreadLocal的myNode变量指向自己的节点。所以获取锁时，使用了每次new一个新的Node，并设置给线程的方式，避免unlock时对node的操作影响兴许节点的状态。也能够使线程多次获得锁。这里能够考虑像CLHLock那样。维护两个ThreadLocal的引用，释放锁时把myNode的引用指向已经不使用的前驱节点。这样避免无谓的new操作。

package com.zc.lock;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 时限队列锁，支持tryLock超时操作
 * QNode维护一个指针preNode指向前一个节点。当preNode == AVAILABLE表示已经释放锁。
当preNode == null表示等待锁
 * tail维护一个虚拟链表,通过tail.getAndSet方法获得前一个节点,并在前一个节点自旋,当释放锁时前一个节点的preNode == AVAIABLE。自己主动通知后一个节点获取锁
 * 当一个节点超时或者被中断，那么它的前驱节点不为空。兴许节点看到它的前驱节点不为空。而且不是AVAILABLE时，知道这个节点退出了，就会跳过它
 * 当节点获得锁。进入临界区后。它的前驱节点能够被回收
 * **/
public class TimeoutLock implements TryLock{
	// 声明为静态变量，防止被暂时回收
	private static final QNode AVAILABLE = new QNode();
	
	// 原子变量指向队尾
	private AtomicReference<QNode> tail;

	ThreadLocal<QNode> myNode;
	
	public TimeoutLock(){
		tail = new AtomicReference<QNode>(null);
		myNode = new ThreadLocal<QNode>(){
			protected QNode initialValue(){
				return new QNode();
			}
		};
	}
	
	@Override
	public void lock() {
		// 和CLHLock不同。每次新建一个Node，并设置给线程，目的是支持同一个线程能够多次获得锁，而不影响链中其它节点的状态
		// CLHLock不须要每次新建Node是由于它使用了两个指针，一个指向前驱节点。
而前驱节点释放后就能够回收了。
		// CLHLock每次释放锁时设置myNode为失效的前驱节点，也是为了支持同一个线程能够多次获取锁而不影响其它节点
		QNode node = new QNode();
		myNode.set(node);
		QNode pre = tail.getAndSet(node);
		if(pre != null){
			// 在前一个节点自旋。当前一个节点是AVAILABLE时，表示它获得锁
			while(pre.preNode != AVAILABLE){
				
			}
		}
	}

	@Override
	public void unlock() {
		QNode node = myNode.get();
		// CAS操作，假设为true。表示是唯一节点，直接释放即可；否则把preNode指向AVAILABLE
		if(!tail.compareAndSet(node, null)){
			node.preNode = AVAILABLE;
		}
	}
	
	@Override
	//TimeUnit仅仅支持毫秒
	public boolean trylock(long time, TimeUnit unit) throws InterruptedException {
		if(Thread.interrupted()){
			throw new InterruptedException();
		}
		boolean isInterrupted = false;
		long startTime = System.currentTimeMillis();
		long duration = TimeUnit.MILLISECONDS.convert(time, unit);
		// 注意：每次tryLock都要new新的Node，为了同一个线程能够多次获得锁。
假设每一个线程都使用同一个节点，会影响链中其它的节点
		QNode node = new QNode();
		myNode.set(node);
		// 尝试一次获取锁
		QNode pre = tail.getAndSet(node);
		// 第一个节点或者之前的节点都是已经释放了锁的节点, pre==AVAILABLE表示获得了锁
		if(pre == null || pre == AVAILABLE){
			return true;
		}
		// 在给定时间内对preNode自旋
		while((System.currentTimeMillis() - startTime < duration) && !isInterrupted){
			QNode predPreNode = pre.preNode;
			// 表示前一个节点已经释放了锁。设置了preNode域，否则preNode域为空
			if(predPreNode == AVAILABLE){
				return true;
			}
			// 当prePreNode ！= null时，仅仅有两种情况，就是它超时了。或者被中断了。

			// 跳过prePreNode不为空的节点。继续自旋它的下一个节点
			else if(predPreNode != null){
				pre = predPreNode;
			}
			if(Thread.interrupted()){
				isInterrupted = true;
			}
		}
		
		// 超时或者interrupted，都要设置node的前驱节点不为空
		node.preNode = pre;
		
		if(isInterrupted){
			throw new InterruptedException();
		}
		
		return false;
	}
	
	public static class QNode {
		volatile QNode preNode;
	}
	
	public String toString(){
		return "TimeoutLock";
	}

}

TimeoutLock具备全部CLHLock的特性，比方无饥饿，先来先服务的公平性，在多个共享变量上自旋，从而控制合理的缓存一致性流量等等，而且支持了限时操作和中断操作。

使用限时锁时有固定的模板，防止锁被错误使用。

Lock lock = ...;
if (lock.tryLock()) {
     try {
     // manipulate protected state
     } finally {
        lock.unlock();
     }
} else {
    // perform alternative actions
}

相同，我们之前验证锁正确性的測试用例相同对TimeoutLock有效，这里不反复帖代码了。

转载请注明来源: http://blog.csdn.net/iter_zc