锁与CompareAndSwap

这篇文章我主要想总结两个内容，第一是关于锁的，第二是关于非阻塞同步CompareAndSwap的。这两个内容在Java多线程并发中都很重要，下面就直接进入主题吧。

锁

要提到并发，自然就要提到锁，通过使用锁，使得多线程的并发控制变得十分简单。但是付出的代价也很高，只有获取到锁的线程才能够执行代码，而其他线程必须挂起等待直到锁被释放，这期间它不能做任何事情。并且，在线程进行切换的过程中，即一个线程释放锁，另一个线程被调度获得锁并执行代码，也存在着很大的系统开销。然而人们对程序效率的追求并没有止步，程序的响应能不能更快一点、效率能不能更高一点等问题不断的激发着人们的热情。于是就产生了多种不同种类的锁，分别适用于在不同的场景下提高程序的并发效率。下面就先说说乐观锁和悲观锁，乐观锁与悲观锁都是概念上的，它们的区别在于悲观锁假设最坏的情况一定会发生，所以就在每次访问共享资源时都上锁。然而我们知道并不是所有的并发操作都会导致数据不一致，这就导致有些本来可以并发的线程由于不能获取到锁而必须等待，从而降低了并发的效率。乐观锁则与之不同，就像它的名字一样，它以一种乐观的态度去访问共享数据，即它认为对共享数据的修改不会造成冲突，所以访问共享资源的时候并不加锁，如果它对共享资源的修改真的产生了冲突，那么它就会放弃这次修改，然后不断的重试。这种乐观锁的形式在下文中还会具体将到，就是CompareAndSwap。讲完了乐观锁与悲观锁，就接着讲讲读写锁吧。读写锁是一种锁分离技术，它把读锁和写锁分开，读锁可以被多个线程持有，这样读线程就可以并发，而写锁是互斥的，它只能被一个线程持有，并且写锁与读锁也互斥。接下来就是可重入锁了，其实synchronized就是可重入的，可重入锁就是说一个线程获取到锁之后，在该线程内部又要递归的获取锁，如果锁不是可重入的，就会造成死锁，因为它不能获取到已经被自己保持的锁。但是可重入锁则不同，一个线程内部递归的获取锁时，会使锁计数器加一，该线程每释放一个锁，锁的计数器就减一，当锁计数器为零时，锁被完全释放。ReentrantLock就是一个可重入锁的实现，它的使用是显式的，并且要在finally块中释放锁，这点与synchronized使用的内置锁不同，在synchronized代码块中的代码如果抛出了异常，内置锁会被自动释放掉。（这样锁也变成了对象，正是万物皆对象！）它比synchronized更加灵活，它为处理锁的不可用性问题提供了解决方案，比如，可以中断一个正在等待获取锁的线程、或者在线程请求获取一个锁时设置超时时间而避免无限的等待下去。基于ReentrantLock机制，Java并发包中还引入了Condition接口，用来提供与Object类中的wait()，notify()和notifyAll()方法类似的await()，signal()和singalAll()方法。下面是一个使用读写锁实现的对Map的包装，增加了并发性能，请看代码：

import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 使用读写锁包装Map，使它能在多个读线程之间安全共享，并且避免读写、写写冲突。
 * 适用于对另一种Map实现提供并发性更高的访问。但是如果仅仅是需要一个并发的Map，
 * 使用ConcurrentHashMap是一个很好的选择。
 * @author Colin Wang
 * Created on Apr 24, 2015
 */
public class ReadWriteMap<K,V> {

    private final Map<K, V> map;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock readLock = lock.readLock();
    private final Lock writeLock = lock.writeLock();

    // 对传入的Map进行包装
    public ReadWriteMap(Map<K, V> map) {
        this.map = map;
    }

    public V put(K key, V value) {
        // 写操作需要取得写锁
        writeLock.lock();
        try {
            return map.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }

    public V get(K key) {
        readLock.lock();
        try {
            return map.get(key);
        } finally {
            readLock.unlock();
        }
    }
}

非阻塞同步算法

关于非阻塞同步算法，本文主要讲一下CompareAndSwap，即比较和交换。这就是前文提到的乐观锁技术，CompareAndSwap有三个操作数：内存值V，预期值A，新值B。当使用CompareAndSwap时，会先将预期值A与内存值V进行比较，如果相同，则把内存值V替换成新值B，如果不相同，则不进行替换，并返回内存中的实际值。这个语义可以解释为：我认为V的值应该为A，如果是，就把V的值替换成B，如果不是就不替换，并告诉我内存中的实际值。在JDK的原子类中都提供了这种基于乐观锁的CAS操作，而且concurrent包中的很多类也使用了这些原子类。在这些原子类中通过调用sun.misc.Unsafe里面的CAS算法，用CPU指令来实现无锁自增。所以，AtomicLong.incrementAndGet()的自增比使用synchronized这种悲观锁的效率要高很多。下面是使用原子引用实现的一个非阻塞栈，它是线程安全的，但是并不是通过同步来实现的。它使用了基于乐观锁的形式，在多线程并发的情况下，代码top.compareAndSet(oldHead, newHead)只会有一个线程执行成功，而其他线程均失败，并可以选择重试。这样避免了使用悲观锁时线程之间的等待唤醒，提高了并发效率。

import java.util.concurrent.atomic.AtomicReference;

/**
 * 非阻塞栈
 * @author Colin Wang
 * Created on Apr 25, 2015
 */
public class ConcurrentStack<E> {

    // 使用原子引用保存当前栈顶元素的引用
    AtomicReference<Node<E>> top = new AtomicReference<>();

    public void push(E e) {
        // 创建一个新的结点
        Node<E> newHead = new Node<E>(e);
        // 存储旧的栈顶
        Node<E> oldHead;
        do {
            // 获取当前的栈顶
            oldHead = top.get();
            // 新结点的next域指向当前的栈顶
            newHead.next = oldHead;
            // 使用CAS更新当前栈顶，如果失败就进行重试。
            // 如果当前栈顶为oldHead，则更新为newHead，操作成功。
            // 如果当前栈顶值不是oldHead，表示其他线程已经对栈顶进行了修改，操作失败并重试。
        } while (!top.compareAndSet(oldHead, newHead));
    }

    public E pop() {
        Node<E> oldHead;
        Node<E> newHead;
        do {
            // 取出当前栈顶
            oldHead = top.get();
            if (oldHead == null) {
                // 如果当前栈顶为null则返回null
                return null;
            }
            // 新的栈顶指向当前栈顶的下一个元素
            newHead = oldHead.next;
            // 尝试使用新的栈顶替换旧的栈顶，失败则重试
        } while (!top.compareAndSet(oldHead, newHead));
        // 返回当前栈顶的元素值
        return oldHead.value;
    }

    // 栈中的元素
    private static class Node<T> {
        private T value;
        public Node<T> next;

        public Node(T value) {
            this.value = value;
        }
    }
}