首页 > 代码库 > 并发编程实践四:实现正确和高效的锁

并发编程实践四:实现正确和高效的锁

你是否觉得锁是一种很神奇的东西,在并发编程中,你只需要将你的代码加上锁,就能保证代码是线程安全的(当然现实和感觉有很大差别,代码的线程安全是非常复杂的),那么,这些都是怎么做到的呢?当存在大量线程同时竞争锁时,竞争失败的锁会怎么做呢?锁又是怎么保证这一切高效的执行的呢?这篇文章将为你回答这些问题,首先我将介绍怎样实现一个正确的锁,然后介绍高效的锁应该具备的条件,最后将介绍两种常用的队列锁算法:CLH锁和MCS锁。
文中将用到一些原子变量的特性,你可以将原子变量看作加强版的volatile变量,具备volatile的属性,并提供了CAS等原子操作,想了解更多原子变量的知识,可以参考《Java并发编程实践》第15章:原子变量与非阻塞同步机制。下面我们就开始了。

锁通常包含两个方法:lock和unlock,按照下面的方式来使用:

Lock lock = new ReentrantLock();

lock.lock();
try {
	// 临界区
} finally {
	lock.unlock();
}

线程通过调用lock来获取锁,然后进入临界区(锁保护的区域),完成后使用unlock来释放锁。锁可以保证线程对临界区操作的原子性和可见性,原子性保证一次只有一个线程能够进入临界区,可见性保证线程对临界区的所有状态的修改都能被其它线程看到,下面我将详细介绍锁的实现机制,看看锁是怎么保证临界区代码的原子性和可见性的,也就是说,你怎样才能实现一个正确的锁。

实现正确的锁

要实现一个锁,就必然要面临一个问题:当线程无法获取到锁的时候,应该怎么做?通常有两种解决方案:1)自旋等待(线程会继续占用CPU,不断重试,直到获取锁成功);2)阻塞(线程放弃CPU,进入阻塞状态,直到被唤醒,然后重新尝试获取锁)。这两种方式有各自的使用场景,自旋等待适用于临界区执行时间短的场景,因为阻塞会导致线程的上下文切换,而上下文切换的代价(可以参考http://coderplay.iteye.com/blog/1481211)是较高的,而当临界区的执行时间较长时,则使用阻塞更为合适。在有些系统中,使用了这两种方式的结合,先自旋等待一段时间,如果不能获取到锁,线程再进入等待状态。为了简单,本文的例子都采用自旋等待的方式。
下面实现了一个简单的自旋锁SimpleLock,它具有一个state标识锁状态,当state为false时,表示锁空闲,而当state为true时,则表示锁已经被占用:
public class SimpleLock implements Lock {
	private AtomicBoolean state = new AtomicBoolean(false);

	@Override
	public void lock() {
		while (state.getAndSet(true)) {
		}
	}

	@Override
	public void unlock() {
		state.set(false);
	}
}
线程通过lock来获取锁,在lock中,线程不断的设置state的值到true,并返回state的旧值,如果state的旧值为false,则线程获取锁成功,并返回,否则一直自旋等待。
线程通过unlock方法来释放锁,在unlock中,线程将state的值设置到false。
那么,SimpleLock是否是一个正确的锁呢?也就是说,线程是否具备原子性和可见性呢?看下面的证明。

原子性

原子性即互斥,一次只能有一个线程能够通过SimpleLock。在SimpleLock中,初始state为false,因此至少有一个线程可以在lock中竞争成功并通过,在这个线程通过后,state被设置为true,因此其它线程无法再通过,也就是初始有且只有一个线程能通过lock。当线程在unlock中将state设置到false后,lock中竞争的线程任然有且只有一个能够设置得到返回值false,其它线程将继续等待。因此,SimpleLock具有原子性。

可见性

impleLock的使用和最早讲到的锁的使用具有同样的结构:
Lock lock = new SimpleLock();

lock.lock();
try {
	// 临界区
} finally {
	lock.unlock();
}
假定存在线程A和线程B竞争锁,线程A先于线程B成功,则A和B的执行将具备如下的happens-before原则(想了解更多的happens-before原则,看我的博文“Java并发编程5-Java存储模式”),:
 1)线程A执行lock.lock成功happens-before于线程A对临界区状态的修改(程序次序法则);
 2)线程A对临界区状态的修改hanppens-before于线程A执行lock.unlock成功(程序次序法则);
 3)线程A执行lock.unlock成功happens-before于线程B执行lock.lock成功(volatle变量法则);
 4)线程B执行lock.lock成功happens-before于线程B对临界区状态的修改(程序次序法则);
 5)线程B对临界区状态的修改happens-before于线程B调用lock.unlock成功(程序次序法则)。
由此,由“传递性法则”我们可以得出:线程A对临界区状态的修改happens-before于线程B对临界区状态的修改。也就是说,线程A对临界区状态的修改对线程B来说都是可见的。
因此,SimpleLock具备可见性。
至此,我们可以得出结论,SimpleLock具有原子性和可见性。


除了原子性和可见性,锁也应该是无饥饿的,即线程经过有限的等待时间后,总是能够成功获取锁,那么SimpleLock是否是无饥饿的呢?在SimpleLock中,每个线程想要获取锁时,都需要和其它线程竞争,获取锁的机会都是均等的,因此可以说SimpleLock是无饥饿的,但是,SimpleLock却不能保证线程等待的时间,可能很长,也可能很短,因此SimpleLock是不公平的。
另外,SimpleLock也是不可重入的,即一个线程不能多次获取锁(这篇文章都不涉及锁的重入,锁的重入将放到下一篇去讲)。
至此,我们就可以得出结论了,SimpleLock是一个正确的锁。但是,很多时候我们不仅希望锁是正确的,我们也希望锁是高效的,那么,SimpleLock是一个高效的锁吗?下面我就来谈谈怎样实现一个高效的锁。

实现高效的锁

我们先对SimpleLock做一些小的改动,然后来做个对比:
public class CASLock implements Lock {
	private AtomicBoolean state = new AtomicBoolean(false);

	@Override
	public void lock() {
		while (!state.compareAndSet(false, true)) {
		}
	}

	@Override
	public void unlock() {
		state.set(false);
	}
}
CASLock和SimpleLock实际上是等同的,只是在设置state时改用了compareAndSet操作,compareAndSet和getAndSet的区别在于:getAndSet总是将state设置为新值,然后返回旧值,而compareAndSet会先比较旧值,符合的情况下才改用新值,那么,改用compareAndSet到底会给我们带来什么改变呢?在《多处理器编程的艺术》中对两种情况作了对比,随着线程数量的增加,SimpleLock要比CASLock性能降低快许多,下面我们从理论上来分析一下原因。
在现代的多处理器系统结构中,都是通过总线采用共享广播媒介的方式进行通讯的,处理器和存储控制器都可以在总线上广播,但一次只能有一个处理器或者存储控制器在总线上广播。每个处理器都有一个cache,用来存储该处理器感兴趣的数据,处理器对cache的访问要比对内存的访问快很多。
当CPU取一个变量值时,首先查看cache,如果在cache中,则CPU产生一个cache命中,并立即加载这个值,否则,产生一个cache缺失,且必须要到内存或者其它CPU的cache中去找这个值。CPU会在总线上广播变量的地址,其它处理器将接收到广播,如果其中一个处理器的cache中存在那个变量的地址,则它会对广播发出回应,否则,CPU将使用内存中的地址的值。
我们再来看SimpleLock的执行过程,SimpleLock每次尝试都会修改state的值,都需要占用主线,而且,每次修改都会导致其它CPU的cache失效,其它CPU都需要通过主线来获取最新的值,大量的主线争用就产生了,更糟糕的是,线程自旋产生的主线争用还会导致其它线程解锁的阻塞(由于主线被占用)。
而CASLock则要好很多,它并不是每次尝试都会修改state的值,而是在state发生变化时,才会导致所有CPU去内存获取最新的值,这样就减少了一部分的主线占用。但是当state的值一旦发生变化后,任然会导致其它CPU的cache失效,从而导致其它CPU对主线的争用。
从上面的所有理论分析中,我们已经不难得出结论:如果我们想实现高性能的锁,我们需要做到:
最大可能的减少数据竞争
这不仅是实现高性能的锁需要做的,这也是所有高效并发编程需要做的。
至于如何减少数据竞争,有一些比较好的策略,如后退和使用队列等。后退是锁每次竞争失败后都等待一段时间,然后再次尝试获取锁;而队列则是使用一个队列来保存等待的锁列表。现在使用较多的是使用队列的方式,因此这里我只介绍队列锁算法,下面就来看看队列锁是怎样将少数据竞争的。

队列锁

队列锁使用一个队列来保存等待获取锁的线程,队列可以是FIFO队列或者优先级队列,前续节点释放锁只会影响后续节点的状态变化。现在常用的队列锁算法有CLH队列锁和MCS队列锁,AQS中就是用的CLH队列锁的一种变种(详见“并发编程实践二:AbstractQueuedSynchronizer”),下面对它们做一个简单的介绍。

CLH队列锁

CLH队列锁为每个线程分配一个节点,节点中使用一个locked属性用来表明该线程已经获取锁或者正在准备获取锁,该节点后续的节点将循环探测该属性,直到该属性为false才能通过。CLH队列锁使用的是一个“虚拟”队列,它将每个节点对应到每个线程,每个线程使用本地变量保存自己的前续节点,具体看下面的代码:
public class CLHLock implements Lock {
	private AtomicReference<Node> tail = new AtomicReference<>();
	private ThreadLocal<Node> myNode = null;
	private ThreadLocal<Node> prevNode = null;

	public CLHLock() {
		tail.set(new Node());
		myNode = new ThreadLocal<Node>() {
			protected Node initialValue() {
				return new Node();
			}
		};
		prevNode = new ThreadLocal<Node>() {
			protected Node initialValue() {
				return null;
			}
		};
	}

	@Override
	public void lock() {
		Node node = myNode.get();
		node.locked = true;
		Node prev = tail.getAndSet(node);
		prevNode.set(prev);
		while (prev.locked) {

		}
	}

	@Override
	public void unlock() {
		Node node = myNode.get();
		node.locked = false;
		myNode.set(prevNode.get());
	}

	private class Node {
		private volatile boolean locked = false;
	}
}
tail用来记录队列尾,线程调用lock后,获取自己的node,并将node的locked设置为true,表明自己准备或者已经获取锁,然后将自己添加到队列尾,并使用本地变量prev保存前续节点,最后循环探测前续节点的locked,直到前续节点的locked为false;线程调用unlock后,去自己对应的node,然后将node的locked设置为false,表明自己已经释放锁。
prevNode的作用是用于记录每个线程的前续节点,它的作用是在unlock时可以将线程对应的节点设置为自己的前续节点,目的是为了节点重用。因为前续节点已经不会再被使用,而自己对应的节点则还在被后续节点使用,当自己再次使用lock请求锁时,会再次修改自己对应的节点的locked属性,就可能会影响到后续节点的处理,因此需要将自己对应的节点替换掉。当然,你也并不是一定需要这样做,你可以直接将线程对应的节点设置到一个新节点
@Override
public void unlock() {
	Node node = myNode.get();
	node.locked = false;
	myNode.set(new Node());
}
由于Java本身就具有垃圾回收机制,因此这样做也不会存在问题。
CLH队列锁带来的好处就是每个节点都循环探测自己的前续节点的locked属性,当一个节点的locked属性发生变化时,只会影响到后续节点的cache失效,并且CLH队列锁可以做到先来先服务。

MCS队列锁

MCS队列锁也是为每个线程分配一个节点,节点中任然包含一个locked属性,和CLH队列锁不同,MCS队列锁使用一个真正的队列来保存等待线程,因此节点中还包含一个next属性,并且locked属性的含义也不一样,在这里表示该线程是否应该被阻塞,线程将循环探测自己节点的locked属性,直到该属性被前续节点的线程修改为false。
public class MCSLock implements Lock {
	private AtomicReference<Node> tail = new AtomicReference<>(null);
	private ThreadLocal<Node> myNode;

	public MCSLock() {
		myNode = new ThreadLocal<Node>() {
			protected Node initialValue() {
				return new Node();
			}
		};
	}

	@Override
	public void lock() {
		Node node = myNode.get();
		Node prev = tail.getAndSet(node);
		if (prev != null) {
			node.locked = true;
			prev.next = node;
			while (node.locked) {
			}
		}
	}

	@Override
	public void unlock() {
		Node node = myNode.get();
		if (node.next == null) {
			if (tail.compareAndSet(node, null)) {
				return;
			}
			while (node.next == null) { //队列尾发生了改变,必然有新的节点正在或者将要添加进来,因此循环等待
			}
		}
		node.next.locked = false;
		node.next = null;
	}

	private class Node {
		private volatile boolean locked = false;
		private volatile Node next;
	}
}
线程调用lock后,首先获取自己对应的节点node,并将node设置为队列尾,并返回前续节点,如果前续节点不为空,则表明线程应该等待:线程首先将node的locked设置为true,表示自己将被阻塞,然后设置前续节点的next指向node,然后就开始循环直到node的locked属性被设置为false。
线程在调用unlock后,首先获取自己对应的节点node,如果node的next为空,则尝试将队列尾设置到空,如果成功,则说明队列已经为空,则退出;否则则说明队列尾发生了变化,需要等待其它线程设置node的next属性,最后设置node的next的locked属性,并退出。
MCS队列锁和CLH队列锁具有相同的特点,前续节点对状态的改变只会影响到后续的节点,不同点是MCS队列锁是在本地cache自旋等待。

结束语

这篇文章为你介绍了锁的基本原理,并讲述了怎样才能实现一个高效的锁,并介绍了两种常用的队列锁算法:CLH队列锁和MCS队列锁。有了这些基本知识后,我们就可以开始Java的Lock和ReadWriteLock的学习了,在学习了Lock和ReadWriteLock后,我还会对JDK 8中新引入的StampedLock做一些介绍。