线程安全的集合

Posted 2020-06-14

除了.net自带的Concurrent系列的线程安全集合，有时候我们可以有自定义的实现，比如使用锁（lock），但是这使得并发性下降。本文将利用Interlocked类实现线程安全的队列。

首先定义一个辅助类Node，这个节点类将作为队列中的元素。

private class Node
{
    public T Value;
    public Node Next;
    
    public Node(T value)
    {
        Value = value;
    } 
}   

其中，我们定义了节点值属性Value，以及表示下一个节点的引用。

有了这个节点类，我们可以不需要使用.net框架的Queue类就能实现集合。

这个节点类与Haskell的List类型有点相似，比如[x, [x, [x, [x, [x]]]]]。假设给出一个头节点和一个尾节点，这个头节点是[x, [x, [x, [x, [x]]]]]，尾节点的Next节点为null，从头节点开始，通过一级一级的引用，就能获取所有的节点值，直到遇到尾节点的Next节点null为止。

为了实现线程安全的队列，我们此时给出想象中的类以及部分字段的样子，如下

public class InterlockedQueue<T>
{
    // other members
    private Node head;
    private Node tail;
    public InterlockedQueue()
    {
        Node node = new Node(default(T));
        head = tail = node;
    }
}           

注意：我们需要把Node类作为InterlockedQueue类的私有类，才能实现参数化类型。

有了存储队列元素的结构之后，剩下的就是考虑如何Enqueue和Dequeue的线程安全操作了，利用前面提到的Interlocked类很容易实现。

Enqueue方法实现

public void Enqueue(T value)
{
    Node node = new Node(value);
    
    while(true)
    {
        Node tail = this.tail;    // Get current tail node
        Node next = tail.Next;     // Get current tail‘s next node
         
        // must be consistent. If not,  re-enter this while loop
        if (object.ReferenceEquals(tail, this.tail))
        {
            // next node of tail must be null, otherwise, other node is inserted, and it needs to re-enter this while loop
            if (object.ReferenceEquals(next, null)
            {
                // begin to insert this node
                if (object.ReferenceEquals(Interlocked.CompareExchange(ref tail.Next, node, next), next)　　// （1）
                {
                    // if consistent, execute insert operation and then break this while loop
                    Interlocked.CompareExchanged(ref this.tail, node, tail);
                    break;
                }
            }
            else    // tail was not pointing to last node
                    //  try to swing Tail to the next node
                Interlocked.CompareExchange(ref this.tail, next, tail);   
         }
    }
}

主要思想是将tail的Next节点指向新节点node，然后再讲tail指向node节点。

注意：（1）处的object.ReferenceEquals方法主要是用于判断是否已经交换，如果没有交换，那必然这个if判断为false。

Dequeue方法实现

public bool Dequeue(out T value)
{
    Node head;
    Node tail;
    Node next;

    while (true)
    {
        // read head
        head = this.head;
        tail = this.tail;
        next = head.Next;

        // Are head, tail, and next consistent?
        if (Object.ReferenceEquals(this.head, head))
        {
            // is tail falling behind
            if (Object.ReferenceEquals(head.Next, tail.Next))
            {
                // is the queue empty?
                if (Object.ReferenceEquals(next, null))
                {
                    value = default(T);

                    // queue is empty and cannot dequeue
                    return false;
                }

                Interlocked.CompareExchange<Node>(
                    ref this.tail,
                    next.Next,
                    tail);
            }
            else // No need to deal with tail
            {
                // read value before CAS otherwise another deque might try to free the next node
                value = next.Value;

                // try to swing the head to the next node
                if (Interlocked.CompareExchange<Node>(
                    ref this.head,
                    next,
                    head) == head)
                {
                    return true;
                }
            }
        }
    }
}

 源代码参考网络。