无锁队列入门

小咪咪 2022-06-14 09:52 267阅读 0赞

## 基本概念解释 ##

### 自旋锁（spin lock) ###

自旋锁是一种非阻塞锁，如果A线程已经获取自旋锁，B线程获取被A占用的自旋锁的时候不会因为资源不足而进入挂起的状态，相反B线程会不断消耗CPU的时间，不停的尝试获取自旋锁。

### 阻塞锁 ###

如果A线程已经获取了阻塞锁，B线程再次获取锁的时候，会因为资源不满足而进入阻塞态同时该线程不再消耗CPU时间，当A线程释放阻塞锁之后，B线程才会进入就绪态，等待CPU调度。

### 可重入锁 ###

可重入锁又称为递归锁，就是在同一线程上该锁是可重入的，对于不同线程则相当于普通的互斥锁。“重入”意味着获取锁的操作粒度是“线程”， 而不是“调用”。

**在多线程共享资源同步状态的操作中，大多数人都使用过以上三种类型的锁。与基于锁的方案相比较，无锁的算法在设计和实现上都相对复杂，但它们在可伸缩性和活跃性上拥有巨大的优势。原因是无锁算法可以使多个线程在竞争相同的资源上不会发生阻塞同时能够在粒度更细的层次上进行协调，并且极大减少上下文切换、系统调用的开销**

### 无锁算法 ###

无锁算法是用底层的原子机器指令代替锁来确保数据在并发访问的一致性。接下来描述的内容大部分参考**《java并发编程实践》** ——原子变量和非阻塞同步机制（15章）

原子操作指令：

*  Compare And Set：查看内存\*reg里的值是否为oldval，如果是的话，则对其赋值newval。
 *  Fetch And Add：一般用来对变量做+1的原子操作。
 *  Test-and-set：写值到某个内存位置并传回旧值。

在分析无锁队列的实现之前先分析下，基于CAS指令实现一个线程安全的计数器，代码如下：

@ThreadSafe
    public class CasCounter { 
        private SimulatedCAS value;
    
        public int getValue() {
            return value.get();
        }   
    
        public int increment() {
            int v;
            do {
                v = value.get();
            } while (v != value.compareAndSwap(v, v + 1));
            return v + 1;
        }   
    }

无锁的计数器在竞争程度不高时，比基于锁的计数器性能高很多。但同时也需要执行更多的操作和更复杂的控制流程。

理论上，如果其他线程在每次竞争CAS时总是获胜的，那么这个线程每次都会重试，就会发生饥饿的问题。

CAS的主要缺点是：调用者必须自己选择处理竞争问题的解决方案（重试、回退、放弃），而锁中能自动处理竞争问题（线程在获取锁之前将一直处于阻塞）。

#### 无锁队列的链表实现 ####

下面的东西主要来自John D. Valois 1994年10月在拉斯维加斯的并行和分布系统系统国际大会上的一篇论文——《[Implementing Lock-Free Queues][]》。

我们先来看一下进队列用CAS实现的方式：

EnQueue(x) //进队列
    {
        //准备新加入的结点数据
        q = new record();
        q->value = x;
        q->next = NULL;
    
        do {
            p = tail; //取链表尾指针的快照
        } while( CAS(p->next, NULL, q) != TRUE); //如果没有把结点链在尾指针上，再试
    
        CAS(tail, p, q); //置尾结点
    }

我们可以看到，程序中的那个 do- while 的 Re-Try-Loop。就是说，很有可能我在准备在队列尾加入结点时，别的线程已经加成功了，于是tail指针就变了，于是我的CAS返回了false，于是程序再试，直到试成功为止。这个很像我们的抢电话热线的不停重播的情况。

你会看到，为什么我们的“置尾结点”的操作（第12行）不判断是否成功，因为：

1.  如果有一个线程T1，它的while中的CAS如果成功的话，那么其它所有的 随后线程的CAS都会失败，然后就会再循环，
2.  此时，如果T1 线程还没有更新tail指针，其它的线程继续失败，因为tail->next不是NULL了。
3.  直到T1线程更新完tail指针，于是其它的线程中的某个线程就可以得到新的tail指针，继续往下走了。

这里有一个潜在的问题——**如果T1线程在用CAS更新tail指针的之前，线程停掉或是挂掉了，那么其它线程就进入死循环了**。下面是改良版的EnQueue()

EnQueue(x) //进队列改良版
    {
        q = new record();
        q->value = x;
        q->next = NULL;
    
        p = tail;
        oldp = p
        do {
            while (p->next != NULL)
                p = p->next;
        } while( CAS(p.next, NULL, q) != TRUE); //如果没有把结点链在尾上，再试
    
        CAS(tail, oldp, q); //置尾结点
    }

我们让每个线程，自己fetch 指针 p 到链表尾。但是这样的fetch会很影响性能。而通实际情况看下来，99.9%的情况不会有线程停转的情况，所以，更好的做法是，你可以接合上述的这两个版本，如果retry的次数超了一个值的话（比如说3次），那么，就自己fetch指针。

好了，我们解决了EnQueue，我们再来看看DeQueue的代码：

DeQueue() //出队列
    {
        do{
            p = head;
            if (p->next == NULL){
                return ERR_EMPTY_QUEUE;
            }
        while( CAS(head, p, p->next) != TRUE );
        return p->next->value;
    }

**注：关于无锁队列的实现参考[http://coolshell.cn/articles/8239.html][http_coolshell.cn_articles_8239.html]**

### 非阻塞算法中的插入算法 ###

链队列必须支持对头节点和尾节点的快速访问。因此，需要单独的head和tail指针。当插入一个元素的过程中需要更新当前最后一个元素的next指针以及指向tail的指针。

@ThreadSafe
    public class LinkedQueue <E> { 
    
        private static class Node <E> { 
            final E item;
            final AtomicReference<LinkedQueue.Node<E>> next;
    
            public Node(E item, LinkedQueue.Node<E> next) {
                this.item = item;
                this.next = new AtomicReference<LinkedQueue.Node<E>>(next);
            }   
        }   
    
        private final LinkedQueue.Node<E> dummy = new LinkedQueue.Node<E>(null, null);
        private final AtomicReference<LinkedQueue.Node<E>> head
                = new AtomicReference<LinkedQueue.Node<E>>(dummy);
        private final AtomicReference<LinkedQueue.Node<E>> tail
                = new AtomicReference<LinkedQueue.Node<E>>(dummy);
    
        public boolean put(E item) {
            LinkedQueue.Node<E> newNode = new LinkedQueue.Node<E>(item, null);
            while (true) {
                LinkedQueue.Node<E> curTail = tail.get();
                LinkedQueue.Node<E> tailNext = curTail.next.get();
                if (curTail == tail.get()) {
                    if (tailNext != null) {
                        // 队列出入中间状态，推进尾节点 —— A
                        tail.compareAndSet(curTail, tailNext); —— B
                    } else {
                        // 队列处于稳定状态，尝试插入新节点
                        if (curTail.next.compareAndSet(null, newNode)) { —— C
                            // 插入操作成功，尝试推进尾节点
                            tail.compareAndSet(curTail, newNode); —— D
                            return true;
                        }   
                    }   
                }   
            }   
        }   
    }

分析以上代码，得出如下结论：

1.在一个包含多个步骤的更新中，必须找到能够确认中间状态和稳定状态的的标识。本例中，当插入一个新元素的时候，需要更新两个指针。首先，更新当前最后一个元素的next指针，将新节点链接到列表队尾；然后，更新tail指针，指向新的队尾。

*  中间状态——tail->next!= null（A、B操作）
 *  稳定状态—— tail->next == null（C、D操作）

2.一个算法成为非阻塞的算法，必须确保当一个线程失败时不会妨碍其他线程继续执行下去。

在上述的描述中如果D步骤失败，不会影响其他线程继续执行；如果D步骤失败，表示可能在D更新tail指针之前已经有其他线程完成了该操作，这里只需要退出即可。

### ABA问题 ###

ABA：如果另一个线程修改V值假设原来是A，先修改成B，再修改回成A。当前线程的CAS操作无法分辨当前V值是否发生过变化。

关于ABA问题有一个例子：在你非常渴的情况下你发现一个盛满水的杯子，你一饮而尽。之后再给杯子里重新倒满水。然后你离开，当杯子的真正主人回来时看到杯子还是盛满水，他当然不知道是否被人喝完重新倒满。解决这个问题的方案的一个策略是每一次倒水假设有一个自动记录仪记录下，这样主人回来就可以分辨在她离开后是否发生过重新倒满的情况。这也是解决ABA问题目前采用的策略。

**以上参考来自知乎，比较简洁易懂**

链队列插入ABA问题解决方案：在Node中不再只存储单个value, 换成存储value、版本号。这样即使某个Node的value域由A变为B，然后又变为A，版本号也将不同。

## 结论 ##

### 非阻塞算法 ###

一个线程在失败或者挂起不会导致其他线程也失败或挂起，这种算法称为非阻塞算法。

### 无锁算法 ###

算法的每个步骤中都存在某个线程能够执行下去，这种算法称为无锁算法。

**注：如果在算法中仅将CAS用于协调线程之间的操作，并且能够正确地实现，那么他既是一种非阻塞算法，又是一种无锁算法**

[Implementing Lock-Free Queues]: http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.53.8674&rep=rep1&type=pdf
[http_coolshell.cn_articles_8239.html]: http://coolshell.cn/articles/8239.html