相信最近看過(guò)我的文章的朋友對(duì)于Microsoft.Extensions.ObjectPool不陌生；復(fù)用、池化是在很多高性能場(chǎng)景的優(yōu)化技巧，它能減少內(nèi)存占用率、降低GC頻率、提升系統(tǒng)TPS和降低請(qǐng)求時(shí)延。

那么池化和復(fù)用對(duì)象意味著同一時(shí)間會(huì)有多個(gè)線程訪問(wèn)池，去獲取和歸還對(duì)象，那么這肯定就有并發(fā)問(wèn)題。那ObjectPool在涉及多線程訪問(wèn)資源應(yīng)該怎么做到線程安全呢？

今天就帶大家通過(guò)學(xué)習(xí)ObjectPool的源碼聊一聊它是如何實(shí)現(xiàn)線程安全的。

源碼解析

ObjectPool的關(guān)鍵就在于兩個(gè)方法，一個(gè)是Get用于獲取池中的對(duì)象，另外就是Return用于歸還已經(jīng)使用完的對(duì)象。我們先來(lái)簡(jiǎn)單的看看ObjectPool的默認(rèn)實(shí)現(xiàn)DefaultObjectPool.cs類的內(nèi)容。

私有字段

先從它的私有變量開始，下面代碼中給出，并且注釋了其作用：

// 用于存放池化對(duì)象的包裝數(shù)組 長(zhǎng)度為構(gòu)造函數(shù)傳入的max - 1
// 為什么 -1 是因?yàn)樾阅芸紤]把第一個(gè)元素放到 _firstItem中
private protected readonly ObjectWrapper[] _items;

// 池化策略 創(chuàng)建對(duì)象 和 回收對(duì)象的防范
private protected readonly IPooledObjectPolicy<T> _policy;

// 是否默認(rèn)的策略 是一個(gè)IL優(yōu)化 使編譯器生成call 而不是 callvirt
private protected readonly bool _isDefaultPolicy;

// 因?yàn)槌鼗蠖鄶?shù)場(chǎng)景只會(huì)獲取一個(gè)對(duì)象 為了性能考慮 單獨(dú)整一個(gè)對(duì)象不放在數(shù)組中 
// 避免數(shù)組遍歷
private protected T? _firstItem;

// 這個(gè)類是在2.1中引入的，以盡可能地避免接口調(diào)用 也就是去虛擬化 callvirt
private protected readonly PooledObjectPolicy<T>? _fastPolicy;

構(gòu)造方法

另外就是它的構(gòu)造方法，默認(rèn)實(shí)現(xiàn)DefaultObjectPool有兩個(gè)構(gòu)造函數(shù)，代碼如下所示：

/// <summary>
/// Creates an instance of <see cref="DefaultObjectPool{T}"/>.
/// </summary>
/// <param name="policy">The pooling policy to use.</param>
public DefaultObjectPool(IPooledObjectPolicy<T> policy)
    : this(policy, Environment.ProcessorCount * 2)
{
    // 從這個(gè)構(gòu)造方法可以看出，如果我們不指定ObjectPool的池大小
    // 那么池大小會(huì)是當(dāng)前可用的CPU核心數(shù)*2
}

/// <summary>
/// Creates an instance of <see cref="DefaultObjectPool{T}"/>.
/// </summary>
/// <param name="policy">The pooling policy to use.</param>
/// <param name="maximumRetained">The maximum number of objects to retain in the pool.</param>
public DefaultObjectPool(IPooledObjectPolicy<T> policy, int maximumRetained)
{
    _policy = policy ?? throw new ArgumentNullException(nameof(policy));

    // 是否為可以消除callvirt的策略
    _fastPolicy = policy as PooleObjectPolicy<T>;
    // 如上面?zhèn)渥⑺f(shuō) 是否為默認(rèn)策略 可以消除callvirt
    _isDefaultPolicy = IsDefaultPolicy();

    // 初始化_items數(shù)組 容量還剩一個(gè)在 _firstItem中
    _items = new ObjectWrapper[maximumRetained - 1];

    bool IsDefaultPolicy()
    {
        var type = policy.GetType();

        return type.IsGenericType && type.GetGenericTypeDefinition() == typeof(DefaultPooledObjectPolicy<>);
    }
}

Get 方法

如上文所說(shuō)，Get()方法是ObjectPool中最重要的兩個(gè)方法之一，它的作用就是從池中獲取一個(gè)對(duì)象，它使用了CAS近似無(wú)鎖的指令來(lái)解決多線程資源爭(zhēng)用的問(wèn)題，代碼如下所示：

public override T Get()
{
    // 先看_firstItem是否有值
    // 這里使用了 Interlocked.CompareExchange這個(gè)方法
    // 原子性的判斷 _firstItem是否等于item
    // 如果等于那把null賦值給_firstItem
    // 然后返回_firstItem對(duì)象原始的值  反之就是什么也不做
    var item = _firstItem;
    if (item == null || Interlocked.CompareExchange(ref _firstItem, null, item) != item)
    {

        var items = _items;
        // 遍歷整個(gè)數(shù)組
        for (var i = 0; i < items.Length; i++)
        {
            item = items[i].Element;
            // 通過(guò)原子性的Interlocked.CompareExchange嘗試讀取一個(gè)元素
            // 讀取成功則返回
            if (item != null && Interlocked.CompareExchange(ref items[i].Element, null, item) == item)
            {
                return item;
            }
        }

        // 如果遍歷整個(gè)沒(méi)有獲取到元素
        // 那么走創(chuàng)建方法，創(chuàng)建一個(gè)
        item = Create();
    }

    return item;
}

上面代碼中，有一個(gè)點(diǎn)解釋一下Interlocked.CompareExchange(ref _firstItem, null, item) != item，其中!=item，如果其等于item就說(shuō)明交換成功了，當(dāng)前線程獲取到_firstItem元素的期間沒(méi)有其它線程修改_firstItem的值。

Return 方法

Retrun(T obj)方法是ObjectPool另外一個(gè)重要的方法，它的作用就是當(dāng)程序代碼把從池中獲取的對(duì)象使用完以后，將其歸還到池中。同樣，它也使用CAS指令來(lái)解決多線程資源爭(zhēng)用的問(wèn)題，代碼如下所示：

public override void Return(T obj)
{
    // 使用策略的Return方法對(duì)元素進(jìn)行處理
    // 比如 List<T> 需要調(diào)用Claer方法清除集合內(nèi)元素
    // StringBuilder之類的也需要調(diào)用Claer方法清除緩存的字符
    if (_isDefaultPolicy || (_fastPolicy?.Return(obj) ?? _policy.Return(obj)))
    {
        // 先嘗試將歸還的元素賦值到 _firstItem中
        if (_firstItem != null || Interlocked.CompareExchange(ref _firstItem, obj, null) != null)
        {
            var items = _items;
            // 如果 _firstItem已經(jīng)存在元素
            // 那么遍歷整個(gè)數(shù)組空間 找一個(gè)存儲(chǔ)為null的空位將對(duì)象存儲(chǔ)起來(lái)
            for (var i = 0; i < items.Length && Interlocked.CompareExchange(ref items[i].Element, obj, null) != null; ++i)
            {
            }
        }
    }
}

從核心的Get()和Set()方法來(lái)看，其實(shí)整個(gè)代碼是比較簡(jiǎn)單的，除了有一個(gè)_firstItem有一個(gè)簡(jiǎn)單的優(yōu)化，其余沒(méi)有什么特別的復(fù)雜的邏輯。

主要的關(guān)鍵就在Interlocked.CompareExchange方法上，我們?cè)谙挛膩?lái)仔細(xì)研究一下這個(gè)方法。

關(guān)于 Interlocked.CompareExchange

Interlocked.CompareExchange它實(shí)際上是一個(gè)CAS的實(shí)現(xiàn)，也就是Compare And Swap，從名字就可以看出來(lái)，它就是比較然后交換的意思。

從下面的代碼段我們也可以看出來(lái)，它總共需要三個(gè)參數(shù)。其特性就是只有當(dāng)localtion1 == comparand的時(shí)候才會(huì)將value賦值給localtion1，另外吧localtion1的原始值返回出來(lái)，這些操作都是原子性的。

// localtion1 需要比較的引用A
// value 計(jì)劃給引用A 賦的值
// comparand 和引用A比較的引用
public static T CompareExchange<T> (ref T location1, T value, T comparand) 
where T : class;

一個(gè)簡(jiǎn)單的流程如下所示：

簡(jiǎn)單的使用代碼如下所示：

var a = 1;
// a == 1的話就將其置為0
// 判斷是否成功就看返回的值是否為a的原始值
if(Interlocked.CompareExchange(ref a, 0, 1) == 1)
	Console.WriteLine("1.成功");
	
// 現(xiàn)在a已經(jīng)變?yōu)? 這個(gè)交換不會(huì)成功
if(Interlocked.CompareExchange(ref a, 0, 1) == 1)
	Console.WriteLine("2.成功");

結(jié)果如下所示，只有當(dāng)a的原始值為1的時(shí)候，才會(huì)交換成功：

那么Interlocked.CompareExchange是如何做到原子性的？在多核CPU中，數(shù)據(jù)可能在內(nèi)存或者L1、L2、L3中(如下圖所示)，我們?nèi)绾伪ＷC能原子性的對(duì)某個(gè)數(shù)據(jù)進(jìn)行操作？

實(shí)際上這是CPU提供的功能，如果查看過(guò)JIT編譯的結(jié)果，可以看到CompareExchange是由一條叫lock cmpxchgl的匯編指令支撐的。

其中lock是一個(gè)指令前綴，匯編指令被lock修飾后會(huì)成為"原子的"，lock指令有兩種實(shí)現(xiàn)方法：

• 早期 - Pentium時(shí)代（鎖總線），在Pentium及之前的處理器中，帶有lock前綴的指令在執(zhí)行期間會(huì)鎖住總線，使得其它處理器暫時(shí)無(wú)法通過(guò)總線訪問(wèn)內(nèi)存，很顯然，這個(gè)開銷很大。
• 現(xiàn)在 - P6以后時(shí)代（鎖緩存），在新的處理器中，Intel使用緩存鎖定來(lái)保證指令執(zhí)行的原子性，緩存鎖定將大大降低lock前綴指令的執(zhí)行開銷。

現(xiàn)在這里的鎖緩存（Cache Locking）就是用了Ringbus + MESI協(xié)議。

MESI協(xié)議是 Cacheline 四種狀態(tài)的首字母的縮寫，分別是修改（Modified）態(tài)、獨(dú)占（Exclusive）態(tài)、共享（Shared）態(tài)和失效（Invalid）態(tài)。 Cache 中緩存的每個(gè) Cache Line 都必須是這四種狀態(tài)中的一種。

修改態(tài)（Modified），如果該 Cache Line 在多個(gè) Cache 中都有備份，那么只有一個(gè)備份能處于這種狀態(tài)，并且“dirty”標(biāo)志位被置上。擁有修改態(tài) Cache Line 的 Cache 需要在某個(gè)合適的時(shí)候把該 Cache Line 寫回到內(nèi)存中。但是在寫回之前，任何處理器對(duì)該 Cache Line在內(nèi)存中相對(duì)應(yīng)的內(nèi)存塊都不能進(jìn)行讀操作。 Cache Line 被寫回到內(nèi)存中之后，其狀態(tài)就由修改態(tài)變?yōu)楣蚕響B(tài)。

獨(dú)占態(tài)（Exclusive），和修改狀態(tài)一樣，如果該 Cache Line 在多個(gè) Cache 中都有備份，那么只有一個(gè)備份能處于這種狀態(tài)，但是“dirty”標(biāo)志位沒(méi)有置上，因?yàn)樗呛椭鲀?nèi)存內(nèi)容保持一致的一份拷貝。如果產(chǎn)生一個(gè)讀請(qǐng)求，它就可以在任何時(shí)候變成共享態(tài)。相應(yīng)地，如果產(chǎn)生了一個(gè)寫請(qǐng)求，它就可以在任何時(shí)候變成修改態(tài)。

共享態(tài)（Shared），意味著該 Cache Line 可能在多個(gè) Cache 中都有備份，并且是相同的狀態(tài)，它是和內(nèi)存內(nèi)容保持一致的一份拷貝，而且可以在任何時(shí)候都變成其他三種狀態(tài)。

失效態(tài)（Invalid），該 Cache Line 要么已經(jīng)不在 Cache 中，要么它的內(nèi)容已經(jīng)過(guò)時(shí)。一旦某個(gè)Cache Line 被標(biāo)記為失效，那它就被當(dāng)作從來(lái)沒(méi)被加載到 Cache 中。

總得來(lái)說(shuō)，若干個(gè)CPU核心通過(guò)Ringbus連到一起。每個(gè)核心都維護(hù)自己的Cache的狀態(tài)。如果對(duì)于同一份內(nèi)存數(shù)據(jù)在多個(gè)核里都有Cache，則狀態(tài)都為S（Shared）。

一旦有一核心改了這個(gè)數(shù)據(jù)（狀態(tài)變成了M），其他核心就能瞬間通過(guò)Ringbus感知到這個(gè)修改，從而把自己的Cache狀態(tài)變成I（Invalid），并且從標(biāo)記為M的Cache中讀過(guò)來(lái)。同時(shí)，這個(gè)數(shù)據(jù)會(huì)被原子的寫回到主存。最終，Cache的狀態(tài)又會(huì)變?yōu)镾。

關(guān)于MESI協(xié)議更詳細(xì)的信息就不在本文中介紹了，在計(jì)算機(jī)操作系統(tǒng)和體系結(jié)構(gòu)相關(guān)書籍和資料中有更詳細(xì)的介紹。

然后compxchg這個(gè)指令就很簡(jiǎn)單了，和我們之前提到的一樣，比較兩個(gè)地址中的值是否相等，如果相等的話那么就修改。

Interlocked類中的其它方法也是同樣的原理，我們可以看看Add之類的方法，同樣是在對(duì)應(yīng)的操作指令前加了lock指令。

總結(jié)

本文主要是帶大家看了下ObjectPool的源碼，然后看了看ObjectPool能實(shí)現(xiàn)無(wú)鎖線程安全的最大功臣Interlocked.CompareExchange方法；然后通過(guò)匯編代碼了解了一下Interlocked類中的一些方法是如何做到原子性的。

到此這篇關(guān)于從ObjectPool到CAS指令的文章就介紹到這了,更多相關(guān)ObjectPool到CAS指令內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論