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

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

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

源碼解析

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

私有字段

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

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

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

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

// 因?yàn)槌鼗蠖鄶?shù)場景只會獲取一個(gè)對象 為了性能考慮 單獨(dú)整一個(gè)對象不放在數(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的池大小
    // 那么池大小會是當(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 是否為默認(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 方法

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

public override T Get()
{
    // 先看_firstItem是否有值
    // 這里使用了 Interlocked.CompareExchange這個(gè)方法
    // 原子性的判斷 _firstItem是否等于item
    // 如果等于那把null賦值給_firstItem
    // 然后返回_firstItem對象原始的值  反之就是什么也不做
    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;
            // 通過原子性的Interlocked.CompareExchange嘗試讀取一個(gè)元素
            // 讀取成功則返回
            if (item != null && Interlocked.CompareExchange(ref items[i].Element, null, item) == item)
            {
                return item;
            }
        }

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

    return item;
}

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

Return 方法

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

public override void Return(T obj)
{
    // 使用策略的Return方法對元素進(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è)存儲為null的空位將對象存儲起來
            for (var i = 0; i < items.Length && Interlocked.CompareExchange(ref items[i].Element, obj, null) != null; ++i)
            {
            }
        }
    }
}

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

主要的關(guān)鍵就在Interlocked.CompareExchange方法上，我們在下文來仔細(xì)研究一下這個(gè)方法。

關(guān)于 Interlocked.CompareExchange

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

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

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

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

簡單的使用代碼如下所示：

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è)交換不會成功
if(Interlocked.CompareExchange(ref a, 0, 1) == 1)
	Console.WriteLine("2.成功");

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

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

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

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

• 早期 - Pentium時(shí)代（鎖總線），在Pentium及之前的處理器中，帶有lock前綴的指令在執(zhí)行期間會鎖住總線，使得其它處理器暫時(shí)無法通過總線訪問內(nèi)存，很顯然，這個(gè)開銷很大。
• 現(xiàn)在 - P6以后時(shí)代（鎖緩存），在新的處理器中，Intel使用緩存鎖定來保證指令執(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)存中。但是在寫回之前，任何處理器對該 Cache Line在內(nèi)存中相對應(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)志位沒有置上，因?yàn)樗呛椭鲀?nèi)存內(nèi)容保持一致的一份拷貝。如果產(chǎn)生一個(gè)讀請求，它就可以在任何時(shí)候變成共享態(tài)。相應(yīng)地，如果產(chǎn)生了一個(gè)寫請求，它就可以在任何時(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)過時(shí)。一旦某個(gè)Cache Line 被標(biāo)記為失效，那它就被當(dāng)作從來沒被加載到 Cache 中。

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

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

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

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

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

總結(jié)

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

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

最新評論