.NET：通過 CAS 來理解數據庫樂觀並發控制，順便給出無鎖的 RingBuffer。

背景

大多數企業開發人員都理解數據庫樂觀並發控制，不過很少有人聽說過 CAS（我去年才聽說這個概念），CAS 是多線程樂觀並發控制策略的一種，一些無鎖的支持並發的數據結構都會使用到 CAS，本文對比 CAS 和 數據庫樂觀並發控制，以此達到強化記憶的目的。

CAS

CAS = Compare And Swap

多線程環境下 this.i = this.i + 1 是沒有辦法保證線程安全的，因此就有了 CAS，CAS 可以保證上面代碼的線程安全性，但是 CAS 並不會保證 Swap 的成功，只有 Compare 成功了才會 Swap，即：沒有並發發生，即：在我讀取和修改之間沒有別人修改。另外說一點，如果 i 是局部變量，即：i = i + 1，那么這段代碼是線程安全的，因為局部變量是線程獨享的。

不明白 CAS 沒關系，下面通過 CAS 的標准模式 和 一個簡單的示例來理解 CAS。

CAS 的標准模式

偽代碼

 1                     var localValue, currentValue;
 2                     do
 3                     {
 4                         localValue = this.
 5 
 6                         var newValue = 執行一些計算;
 7 
 8                         currentValue = Interlocked.CompareExchange(ref this.value, newValue, localValue);
 9                     } while (localValue != currentValue);

說明

把 this.value 看成是數據庫數據，localValue 是某個用戶讀取的數據，newValue是用戶想修改的值，這里有必要解釋一下 CompareExchange 和 currentValue，它的內部實現代碼是這樣的（想想下面代碼是線程安全的）：

1 var currentValue = this.value
2 if(currentValue == localValue){
3    this.value = newValue;
4 }
5 return currentValue;

CompareExchange  用 sql 來類比就是：update xxx set value = newValue where value = localValue，只不過返回的值不同。通過 CompareExchange 的返回結果我們知道 CAS 是否成功了，即：是否出現並發了，即：是否在我讀取和修改之間別人已經修改過了，如果是，可以選擇重試。

累加示例

CAS 代碼

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;

namespace InterlockStudy
{
    class ConcurrentIncrease
    {
        public static void Test()
        {
            var sum = 0;

            var tasks = new Task[10];
            for (var i = 1; i <= 10; i++)
            {
                tasks[i - 1] = Task.Factory.StartNew((state) =>
                {
                    int localSum, currentSum;
                    do
                    {
                        localSum = sum;

                        Thread.Sleep(10);
                        var value = (int)state;
                        var newValue = localSum + value;

                        currentSum = Interlocked.CompareExchange(ref sum, newValue, localSum);
                    } while (localSum != currentSum);
                }, i);
            }

            Task.WaitAll(tasks);

            Console.WriteLine(sum);
        }
    }
}

數據庫並發代碼

        public static void Test13()
        {
            var tasks = new Task[10];
            for (var i = 1; i <= 10; i++)
            {
                tasks[i - 1] = Task.Factory.StartNew((state) =>
                {
                    int localSum, result;
                    do
                    {
                        using (var con = new SqlConnection(CONNECTION_STRING))
                        {
                            con.Open();
                            var cmd = new SqlCommand("select sum from Tests where Id = 1", con);
                            var reader = cmd.ExecuteReader();
                            reader.Read();
                            localSum = reader.GetInt32(0);

                            System.Threading.Thread.Sleep(10);
                            var value = (int)state;
                            var newValue = localSum + value;

                            cmd = new SqlCommand("update Tests set sum = " + newValue + " where sum = " + localSum + "", con);
                            result = cmd.ExecuteNonQuery();
                        }
                    } while (result == 0);
                }, i);
            }

            Task.WaitAll(tasks);
        }
    }

說明

我們發現 CAS 版本的代碼和數據庫版本的代碼出奇的相似，數據庫的CAS操作是通過 update + where 來完成的。

寫着玩的 RingBuffer

代碼

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;

namespace InterlockStudy
{
    internal class Node<T>
    {
        public T Data { get; set; }

        public bool HasValue { get; set; }
    }

    class RingBuffer<T>
    {
        private readonly Node<T>[] _nodes;
        private long _tailIndex = -1;
        private long _headIndex = -1;
        private AutoResetEvent _readEvent = new AutoResetEvent(false);
        private AutoResetEvent _writeEvent = new AutoResetEvent(false);

        public RingBuffer(int maxSize)
        {
            _nodes = new Node<T>[maxSize];

            for (var i = 0; i < maxSize; i++)
            {
                _nodes[i] = new Node<T>();
            }
        }

        public void EnQueue(T data)
        {
            while (true)
            {
                if (this.TryEnQueue(data))
                {
                    _readEvent.Set();
                    return;
                }

                _writeEvent.WaitOne();
            }

        }

        public T DeQueue()
        {
            while (true)
            {
                T data;
                if (this.TryDeQueue(out data))
                {
                    _writeEvent.Set();
                    return data;
                }

                _readEvent.WaitOne();
            }

        }

        public bool TryEnQueue(T data)
        {
            long localTailIndex, newTailIndex, currentTailIndex;
            do
            {
                localTailIndex = _tailIndex;

                if (!this.CanWrite(localTailIndex))
                {
                    return false;
                }

                newTailIndex = localTailIndex + 1;

                if (_nodes[newTailIndex % _nodes.Length].HasValue)
                {
                    return false;
                }

                currentTailIndex = Interlocked.CompareExchange(ref _tailIndex, newTailIndex, localTailIndex);
            }
            while (localTailIndex != currentTailIndex);

            _nodes[newTailIndex % _nodes.Length].Data = data;
            _nodes[newTailIndex % _nodes.Length].HasValue = true;

            return true;
        }

        public bool TryDeQueue(out T data)
        {
            long localHeadIndex, newHeadIndex, currentHeadIndex;
            do
            {
                localHeadIndex = _headIndex;

                if (!this.CanRead(localHeadIndex))
                {
                    data = default(T);
                    return false;
                }

                newHeadIndex = localHeadIndex + 1;
                if (_nodes[newHeadIndex % _nodes.Length].HasValue == false)
                {
                    data = default(T);
                    return false;
                }

                currentHeadIndex = Interlocked.CompareExchange(ref _headIndex, newHeadIndex, localHeadIndex);
            }
            while (localHeadIndex != currentHeadIndex);

            data = _nodes[newHeadIndex % _nodes.Length].Data;
            _nodes[newHeadIndex % _nodes.Length].HasValue = false;

            return true;
        }

        private bool CanWrite(long localTailIndex)
        {
            return localTailIndex - _headIndex < _nodes.Length;
        }

        private bool CanRead(long localHeadIndex)
        {
            return _tailIndex - localHeadIndex > 0;
        }
    }
}

備注

倉促成文，如果有必要可以再寫篇文章，希望大家多批評。