減小鎖定的粒度：C#實現基於關鍵字（key）的鎖定

問題描述

最近需要實現一個API，方法簽名（的抽象版本）類似於

void Update(string id)

API將在多線程環境下被調用，需滿足：

1. 如果多個調用線程傳入相同的id，則它們必須被串行化——一個線程工作，其他線程阻塞，前一個線程調用完畢后，后一個線程才開始工作，依此類推。
2. 若傳入的id不同，則各線程可並行執行。

場景與數據庫的行鎖定非常相似——鎖定對於更新相同的行的多個請求是互斥的，而更新不同的行則可同時進行。
不過這回我們沒有數據庫的幫忙，同時，程序非常的小（其實是客戶端程序），所以我們希望解決方案也非常小巧。

 

基本思路

說道多線程串行化，立刻想到的就是鎖，但是如果簡單的 lock (someGlobalObject) 會時所有的線程串行化，這不滿足需求。
我們仍然需要鎖，不過鎖的作用范圍更小，於是，需求被轉化為小粒度鎖定的實現，這個鎖范圍滿足：

1. 相同的id共享同一個鎖對象。
2. 不同的id使用不同的鎖對象。

 

實現

思路確定了，進入實現階段。
這里為了便於測試，定義一個接口，實際的場景中並不需要這個接口。

interface IKeyLockEngine
{
    void Invoke(string key, Action act);
}

實現方案1：使用字典記錄id與鎖對象

“id -> 鎖對象”的映射場景很容易讓人想到字典（哈希表）——使用一個Dictionary存放正在被使用的id（作為key）和鎖對象（作為value），若已經沒有線程使用某一id調用API，則從字典中移除該id。
對於每次傳入的id做一次檢驗，將獲得以下兩種情況：

1. id不存在於字典中——沒有線程正在使用該id調用API，為該id分配一個鎖對象，並將id寫入字典；
2. id存在於字典中——已有線程正在使用該id調用API，從字典中取出鎖對象並使用之；

那么，在有多個線程使用同一個id的情況下，如何知道何時需要從字典中移除該id呢？這里引入一個計數器來解決。
此方案的實現代碼如下：

public class DictionaryBasedKeyLockEngine : IKeyLockEngine
{
    private static readonly object SyncRoot = new object();
    private static readonly Dictionary<string, LockUnit> Locks = new Dictionary<string, LockUnit>();

    public void Invoke(string key, Action act)
    {
        LockUnit lockUnit;

        lock (SyncRoot)
        {
            if (Locks.TryGetValue(key, out lockUnit))
            {
                lockUnit.WaitCounter++;
            }
            else
            {
                lockUnit = new LockUnit();
                Locks.Add(key, lockUnit);
            }
        }

        try
        {
            Monitor.Enter(lockUnit);
            act();
        }
        finally
        {
            lock (SyncRoot)
            {
                lockUnit.WaitCounter--;
                if (lockUnit.WaitCounter == 0)
                    Locks.Remove(key);
            }
            Monitor.Exit(lockUnit);
        }
    }

    private class LockUnit
    {
        public int WaitCounter;
    }
}

上面使用了內部類LockUnit作為鎖定對象的類型，並保持一個計數器WaitCounter，其記錄了當前使用該鎖對象的線程的數量，當計數器歸0，就是從字典里移除id的時候了。
因為字典數據是各線程共享的，為了能安全的操作字典，需要一個額外的鎖對象SyncRoot，因此實際上有兩層的鎖定。

實現方案2：互斥體

另一種實現方案使用了.net Framework提供的互斥體 System.Threading.Mutex，利用其可命名的特性，將id作為互斥體的名稱，很好的實現了id到鎖對象的映射。

public class MutexBasedKeyLockEngine : IKeyLockEngine
{
    private static readonly string NameHeader = Guid.NewGuid().ToString("N");

    public void Invoke(string key, Action act)
    {
        var m = new Mutex(false, NameHeader + key);
        try
        {
            m.WaitOne();
            act();
        }
        finally
        {
            m.ReleaseMutex();
        }
    }
}

因為互斥體是在整個操作系統中有效的，作用域非常大，為了避免key與本實現外部所注冊的互斥體沖突，定義了一個Guid（幾乎不會重復）作為互斥體名稱的前綴，以避免此問題。

此方案還有一個問題：互斥體的名稱必須是字符串，而前面使用字典的方案中，key可以是任意類型，容易將string類型的key改為泛型類型，從而擴展該實現的使用范圍。互斥體命名則不能做此擴展，因為我們無法確保關鍵字的類型總是按要求重載了ToString方法。

實現方案3：自旋鎖

此方案由邊城浪補充。利用高性能的CAS操作將鎖的粒度變小，收錄如下：

public class SpinLockEngine : IKeyLockEngine
{
    private const int LockeCount = 0x12fd; //素數，減少hash沖突，值越大沖突概率越小，但占用內存越大
    private static readonly int[] Locks = new int[LockeCount];

    public void Invoke(string key, Action act)
    {
        int index = (key.GetHashCode() & 0x7fffffff) % LockeCount;

        // 嘗試0變1,進入對應index的臨界狀態;
        while (Interlocked.CompareExchange(ref Locks[index], 1, 0) == 1)
        {
            Thread.Sleep(1);
            ////可也以計數方式.每X次嘗試失敗則睡眠1,否則睡眠0
            //Thread.Sleep(((++count) | 0x000f) == 0 ? 1 : 0);
        }

        try
        {
            act();
        }
        finally
        {
            Thread.VolatileWrite(ref Locks[index], 0);
        }
    }
}

該方案下，若兩個的key的GetHashCode結果與素數的模數相同，則兩個key互斥，即使兩個key不想等——目前的應用場景下這是可以接受的。

其他方案：

使用旗語 System.Threading.Semaphore，與互斥體相似，不過還多一個功能，可以控制並發的數量，因為應用場景下沒有此要求，在此便不再討論。

 

性能分析

我們來比較上述兩種方案的性能，重點比較id的重復率對於性能的影響，測試代碼如下，代碼中使用了老趙的性能計數器CodeTimer。

static void Main()
{
    var ran = new Random();
    var keyRange = 100; //控制id重復的概率，值越大重復的概率越小
    var keys = new string[100000];
    for (int i = 0; i < keys.Length; i++)
    {
        keys[i] = ran.Next(keyRange).ToString();
    }

    Action act = () => Thread.Sleep(TimeSpan.FromMilliseconds(0.1));

    Console.WriteLine("keyRange={0}", keyRange);
    CodeTimer.Initialize();
    CodeTimer.Time("mutex", 1, () => Perform(new MutexBasedKeyLockEngine(), keys, act, 10));
    CodeTimer.Time("dictionary", 1, () => Perform(new DictionaryBasedKeyLockEngine(), keys, act, 10));
    CodeTimer.Time("spinlock", 1, () => Perform(new SpinLockEngine(), keys, act, 10));

    Console.ReadKey();
}

static void Perform(IKeyLockEngine keyLockEngine, string[] keys, Action act, int threadCount)
{
    var threads = new List<Thread>();
    for (int i = 0; i < threadCount; i++)
    {
        var tmp = i;
        var t = new Thread(() =>
        {
            for (int j = tmp; j < keys.Length; j += threadCount)
            {
                keyLockEngine.Invoke(keys[j], act);
            }
        });
        threads.Add(t);
    }
    threads.ForEach(x => x.Start());
    threads.ForEach(x => x.Join());
}

測試代碼中使用變量keyRange控制隨機數的生成范圍，keyRange越小，隨機數的取值范圍就越小，生成的關鍵字重復的概率就越大，下面是不同的keyRange下的測試結果。

keyRange=10
mutex
        Time Elapsed:   1,170ms
dictionary
        Time Elapsed:   3,305ms
spinlock
        Time Elapsed:   65ms
====================
keyRange=10000
mutex
        Time Elapsed:   1,277ms
dictionary
        Time Elapsed:   179ms
spinlock
        Time Elapsed:   45ms
====================
keyRange=10000000
mutex
        Time Elapsed:   4,900ms
dictionary
        Time Elapsed:   189ms
spinlock
        Time Elapsed:   56ms

從測試結果中，可以看到，關鍵字的重復率較高時，使用互斥體的方案竟比之使用字典+Moniter的方案的耗時更少；反之，則使用字典的方案耗時更少。
這個結果在使用互斥體的方案上是容易理解的，因為互斥體的創建和銷毀開銷較大，重復率越高則創建/銷毀的互斥體越少，開銷也就越少。
那為何使用字典的方案在關鍵字重復率較高時性能下降了呢？經測試，在多個線程請求鎖時，Moniter.Enter方法比單線程請求鎖時花費的時間更多，這就需要從Moniter的實現原理上去理解了。
自旋鎖的性能表現極好——它的操作更接近底層。

 

結論

若嚴格要求具有不同的key的操作可並行執行，使用字典的方案；在允許不同的key有小概率互斥的情況下，自旋鎖的方案具有最佳的表現。

 

寫在后面：這篇文章的標題怎么取非常讓人糾結，一下子找不到一句合適的話描述該問題，各位會怎么做呢？