Java多线程奇幻之旅——CAS算法实现线程安全

前言

对于线程安全，咱们有说不尽的话题。大多数保证线程安全的方法是添加各类类型锁，使用各类同步机制，用限制对共享的、可变的类变量并发访问的方式来保证线程安全。文本从另外一个角度，使用“比较交换算法”（CompareAndSwap）实现一样的需求。咱们实现一个简单的“栈”，并逐步重构代码来进行讲解。
本文通俗易懂，不会涉及到过多的底层知识，适合初学者阅读（言外之意是各位大神能够绕道了）。

旅程开始

1.先定个小目标，实现一个“栈”

“栈”（stack）是你们常用的抽象数据类型（啥？！不知道，请自行百度）。“栈”知足“后进先出”特性。咱们用链表数据结构完成一个简单的实现：

public class Stack<E> {
    //链表结构头部节点
    private Node<E> head;
    /**
     * 入栈
     * @param item
     */
    public void push(E item) {
        //为新插入item建立一个新node
        Node<E> newHead = new Node<>(item);
        if(head!=null){
            //将新节点的下一个节点指向原来的头部
            newHead.next = head;
        }
        //将头部指向新的节点
        head=newHead;
    }
    /**
     * 出栈
     * @return
     */
    public E pop() {
        if(head==null){
            //当前链表为空
            return null;
        }
        //暂存当前节点。
        Node<E> oldHead=head;
        //将当前节点指向当前节点的下一个节点
        head=head.next;
        //从暂存的当前节点记录返回数据
        return oldHead.item;
    }
    /**
     * 链表中的节点
     * @param <E>
     */
    private static class Node<E> {
        //节点保存的数据
        public final E item;
        //指向下一个链表中下一个节点
        public Node<E> next;
        public Node(E item) {
            this.item = item;
        }
    }
}

代码使用链表数据结构实现“栈”，在Stack中维护一个链表的“头部节点”，经过对头部节点的操做完成入栈和出栈操做。
咱们运行代码测试一下：

public static void main(String[] args) {
    Stack<Integer> stack=new Stack<>();
    for (int i = 0; i < 3; i++) {
        //入栈一、二、3
        stack.push(i+1);
    }
    for (int i = 0; i < 3; i++) {
        //出栈三、二、1
        System.out.println(stack.pop());
    }
}

结果为：

3
2
1

咱们使用入栈方法向Stack插入一、二、3，使用出栈方法打印为三、二、1，符合预期。

2.让多线程捣捣乱

前面咱们已经测试过咱们的方法，符合咱们对Stack功能的预期，那是否是任何状况先咱们的“栈”都能正常工做呢？

咱们运行以下代码：

public static void main(String[] args) {
    Stack<Integer> stack=new Stack<>();
    int max=3;
    Thread[] threads=new Thread[max];
    for (int i = 0; i < max; i++) {
        int temp=i;
        //入栈一、二、3
        Thread thread=new Thread(new Runnable() {
            @Override
            public void run() {
                stack.push(temp+1);
            }
        });
        thread.start();
        threads[temp]=thread;
    }
    //等待全部线程完成。
    for (int i = 0; i < max; i++) {
        try {
            threads[i].join();
        } catch (InterruptedException e) {
        }
    }

    for (int i = 0; i < max; i++) {
        //出栈三、二、1
        System.out.println(stack.pop());
    }
}

你可能运行了不少次，每次运行时除了打印顺序（三、二、1或二、三、1或一、二、3）有变化以外也没有发现其余异常，你可能会说打印顺序变化很正常呀，由于咱们的将入栈操做放到异步线程中操做，三个线程的执行过程由系统调度，因此入栈操做的内容天然每次都有可能不一样。
好吧，你说的没错，至少从大量运行的结果上看是这样的，可是这就是多线程编程的奇（tao）幻(yan)之处，也许你运行一次没有问题，两次没有问题，一万次也没有问题，可是终有一次你会获得那个意想不到的结果（你也不想获得，由于那是bug）。这就像一个“黑天鹅事件”，小几率可是必定会发生，且发生后对你的系统影响不堪设想。
下面让我带你看看如何获得意料以外的结果：

咱们使用调试模式运行上面的程序在Stack中push()方法第一行打一个断点，而后按照表格中的顺序切换不一样的线程以单步调试（step over）方式运行run方法中的每一步，直到遇到Resume。

执行顺序	thread-0	thread-1	thread-2
1	Node<E> newHead = new Node<>(item);	--	--
2	head=newHead;	--	--
3	(Resume)	--	--
4	--	Node<E> newHead = new Node<>(item);	--
5	--	--	Node<E> newHead = new Node<>(item);
6	--	newHead.next = head;	--
7	--	--	newHead.next = head;
8	--	head=newHead;	--
9	--	--	head=newHead;
10	--	(Resume)
11	--	--	(Resume)

当你再次看到打印结果，你会发现结果为三、一、null，“黑天鹅”出现了。

异常结果是如何产生的？
当thread-0执行到顺序3时，head表示的链表为node(1)。
当thread-1执行到顺序10时，head表示的链表为node(2)->node(1)。
当thread-2执行到顺序11时，head表示的链表为node(3)->node(1)。
当三个线程都执行完毕以后，head的最终表示为node(3)->node(1)，也就是说thread-2将thread-1的执行结果覆盖了。

语句newHead.next = head;是对头部节点的读取。语句head=newHead;是对头部节点的写入操做。这两条语句组成了一个“读取——设置——写入”语句模式（就像n=n+1）。
若是一个线程执行了共享头部变量读取语句，切换其余线程执行了修改共享变量的值，再切回到第一个线程后，第一个线程中修改头部结点的数据就不是最新的数据为依据的，因此修改以后其余线程的修改就被覆盖了。
只有保证这两条语句及中间语句以原子方式执行，才能避免多线程覆盖问题。
你们能够任意调整代码中读取头部节点和写入头部节点的调试顺序，制造多线程交错读写观察不一样的异常结果。

为何咱们直接执行没法看到异常结果呢？
由于咱们的run方法很简单，在CPU分配的时间片内能运行完，没有出如今不一样的运行周期中交错运行的状态。因此咱们才要用调试模式这种交错运行。

为何上文中我说过这种异常必定会发生？
缘由在于咱们在Stack类中对共享的、可变的变量head进行的多线程读写操做。

怎么才能保证类Stack在多线程状况下运行正确？
引用一段《JAVA并发编程实践》中的话：

不管什么时候，只要有多于一个的线程访问给定的状态变量，并且其中某个线程会写入该变量，此时必须使用同步来协调线程对该变量的访问。

好吧，看来咱们必须采用“同步”方法了，来保障咱们的Stack类在多线程并行和单线程串行的状况下都有正确的结果，也就是说将Stack变成一个线程安全的类。

3.让你捣乱，请家长！

既然多线程总来捣乱，咱们就请他的家长，让家长管管他，守守规矩，不在捣乱。
咱们已经知道了Stack类问什么不能再多线程下正确的运行的缘由，全部咱们要限制多线程对Stack类中head变量的并发写入，Stack方法中push()和pop()方法都会对head进行写操做，因此要限制这两个方法不能多线程并发访问，因此咱们想到了synchronized关键字。

程序重构：

public class SynchronizedStack<E> {
    //链表结构头部节点
    private Node<E> head;
    /**
     * 入栈
     * @param item
     */
    public synchronized void push(E item) {
        //为新插入item建立一个新node
        Node<E> newHead = new Node<>(item);
        if(head!=null){
            //将新节点的下一个节点指向原来的头部
            newHead.next = head;
        }
        //将头部指向新的节点
        head=newHead;
    }
    /**
     * 出栈
     * @return
     */
    public synchronized E pop() {
        if(head==null){
            //当前链表为空
            return null;
        }
        //暂存当前节点。
        Node<E> oldHead=head;
        //将当前节点指向当前节点的下一个节点
        head=head.next;
        //从暂存的当前节点记录返回数据
        return oldHead.item;
    }
    /**
     * 链表中的节点
     * @param <E>
     */
    private static class Node<E> {
        //节点保存的数据
        public final E item;
        //指向下一个链表中下一个节点
        public Node<E> next;
        public Node(E item) {
            this.item = item;
        }
    }
}

将Stack类替换为SynchronizedStack类的测试方法。

public static void main(String[] args) {
        SynchronizedStack<Integer> stack=new SynchronizedStack<>();
        int max=3;
        Thread[] threads=new Thread[max];
        for (int i = 0; i < max; i++) {
            int temp=i;
            //入栈一、二、3
            Thread thread=new Thread(new Runnable() {
                @Override
                public void run() {
                    stack.push(temp+1);
                }
            });
            thread.start();
            threads[temp]=thread;
        }
        //等待全部线程完成。
        for (int i = 0; i < max; i++) {
            try {
                threads[i].join();
            } catch (InterruptedException e) {
            }
        }

        for (int i = 0; i < max; i++) {
            //出栈三、二、1
            System.out.println(stack.pop());
        }
    }

咱们再次运行第二章为多线程准备的测试方法，发现当执行一个线程的方法时，其余线程的方法均被阻塞，只能等到第一个线程方法执行完成以后才能执行其余线程方法。
咱们只不过是在push()和pop()方法上加入了synchronized 关键字，就将这两个方法编程了同步方法，在多线程并发的状况下也如同单线程串行调用通常，方法再不能在线程间交替运行，也就不能对head变量作并发更改了，这样修改的Stack类就是线程安全的了。

除了synchronized关键字，还有其余的方式实现加锁吗？
除了synchronized关键字还可使用java.util.concurrent.locks包中各类锁来保证同步，可是大概思路都是相同的，都是使用阻塞其余线程的方式在达到防止并发写入的目的。

阻塞线程是否会影响执行效率？
若是和不加经过的“栈”类相比，在多线程执行的以后效率必定会有影响，由于同步方法限制了线程之间的并发性，可是为了保证“栈”类的在多线程环境时功能正确，咱们不得不作出效率和正确性的权衡。

必需要对整个方法加上锁吗？
咱们上面已经分析了须要加锁的范围，只要保证读取头部节点和写入头部节点之间的语句原子性就能够。因此咱们能够这样执行。

/**
     * 入栈
     *
     * @param item
     */
    public void push(E item) {
        //为新插入item建立一个新node
        Node<E> newHead = new Node<>(item);
        synchronized (this) {
            if (head != null) {
                //将新节点的下一个节点指向原来的头部
                newHead.next = head;
            }
            //将头部指向新的节点
            head = newHead;
        }
    }

    /**
     * 出栈
     *
     * @return
     */
    public E pop() {
        synchronized (this) {
            if (head == null) {
                //当前链表为空
                return null;
            }
            //暂存当前节点。
            Node<E> oldHead = head;
            //将当前节点指向当前节点的下一个节点
            head = head.next;
            //从暂存的当前节点记录返回数据
            return oldHead.item;
        }
    }

经过synchronized 块实现，由于方法比较简单，因此也没有很明显的缩小加锁范围。

除了加锁的方式，是否还有其余方式？
固然，咱们还有无锁化编程来解决线程之间同步的问题。这就是下面要介绍的比较交换算法。

4.换个思路，乐观一点

加锁实现线程同步的方式是预防性方式。不管共享变量是否会被并发修改，咱们都只容许同一时刻只有一个线程运行方法来阻止并发发生。这就至关于咱们假设并发必定会发生，因此比较悲观。
如今咱们换一种思路，乐观一点，不要假设对变量的并发修改必定发生，这样也就不用对方法加锁阻止多线程并行运行方法了。可是一旦发生了并发修改，咱们想法发解决就是了，解决的方法就是将这个操做重试一下。

继续重构“栈”代码：

public class TreiberStack<E> {
    private AtomicReference<Node<E>> headNode = new AtomicReference<>();
    public void push(E item) {
        Node<E> newHead = new Node<>(item);
        Node<E> oldHead;
        do {
            oldHead = headNode.get();
            newHead.next = oldHead;
        } while (!headNode.compareAndSet(oldHead, newHead));
    }
    public E pop() {
        Node<E> oldHead;
        Node<E> newHead;
        do {
            oldHead = headNode.get();
            if (oldHead == null)
                return null;
            newHead = oldHead.next;
        } while (!headNode.compareAndSet(oldHead, newHead));
        return oldHead.item;
    }
    private static class Node<E> {
        public final E item;
        public Node<E> next;

        public Node(E item) {
            this.item = item;
        }
    }
}

这个就是大名鼎鼎的Treiber Stack，我也只是作了一次代码的搬运工。
咱们来看看TreiberStack和咱们前面的Stack有什么不一样。

首先关注第一行：

private AtomicReference<Node<E>> headNode = new AtomicReference<>();

咱们用了一个AtomicReference类存储链表的头部节点，这个类能够获取存储对象的最新值，而且在修改存储值时候采用比较交换算法保证原子操做，具体你们能够自行百度。

而后重点关注pop()和push()方法中都有的一个代码结构：

//略...
do {
    oldHead = headNode.get();
    //略...
     
} while (!headNode.compareAndSet(oldHead, newHead));
//略...

咱们AtomicReference中get()方法最新的获取头部节点，而后调用AtomicReference中compareAndSet()将设置新头部节点，若是当前线程执行这两端代码的时候若是有其余已经修改了头部节点的值，'compareAndSet()'方法返回false ,代表修改失败，循环继续，不然修改为功，跳出循环。

这样一个代码结构和synchronized关键字修饰的方法同样，都保证了对于头部节点的读取和写入操做及中间代码在一个线程下原子执行，前者是经过其余线程修改过就重试的方式，后者经过阻塞其余线程的方式，一个是乐观的方式，一个是悲观的方式。

你们能够按照前面的例子本身写测试方法测试。

后记

咱们经过对“栈”的一步一步代码重构，逐步介绍了什么是线程安全及保证线程安全的各类方法。这里须要说明一点，对于一个类来讲，是否须要支持线程安全是由类的使用场景决定，不是有类所提供的功能决定的，若是一个类不会被应用于多线程的状况下也就无需将他转化为线程安全的类。

关于CAS特色等更多内容鉴于本文篇幅有限，我会另文再续。

---

参考

《JAVA并发编程实践》