文本由CocoaChina譯者candeladiao（GitHub）翻譯

作者：Mike Ash（Blog GitHub）

原文：Friday Q&A 2015-09-04: Let's Build dispatch_queue

---

Grand Central Dispatch是蘋果過去幾年創造出來的非常強大的API，在Let's Build系列的最新一期中，我們將探究dispatch_queue基礎功能的重新實現。該主題是Rob Rixr提議的。

概述

dispatch queue是一個工作隊列，其背後是一個全局的線程池。特別是，提交到隊列的任務會在後台線程異步執行。所有線程共享同一個後台線程池，這使得系統更有效率。

這也是我將要模仿的API的精髓部分。GCD還提供了很多精心設計的功能，為了簡單起見，本文將把它們都略過。比如線程池的線程數量會根據待完成的任務數和系統CPU的使用率動態作調整。如果你已經有一堆任務占滿了CPU，然後再扔給它另一個任務，GCD不會再創建另外的工作線程，因為CPU已經被100%占用，再執行別的任務只會更低效。這裡我會寫死線程數而不做模擬動態調整。同時我還會忽略並發隊列的目標隊列和調度屏障功能。

我們目標是聚焦於dispatch queue的真髓：能串行、能並行、能同步、能異步以及共享同一個線程池。

編碼

和以往一樣，今天文章的代碼可以在GitHub上找到：https://github.com/mikeash/MADispatchQueue

如果你想讀的過程中自己探索，以上是所有代碼。

接口

GCD是基於C語言的 API。雖然最新的系統版本中GCD對象已經轉成了Objective-C對象，但API仍保持純C接口(加了block擴展)。這對實現底層接口是好事，GCD提供了出色而簡單的接口，但對我個人而言，我更喜歡用Objective-C來實現。

Objective-C類名稱為MADispatchQueue，包含四個調用方法：

1.獲取全局共享隊列的方法。GCD有多個不同優先級的全局隊列，出於簡單考慮，我們在實現中保留一個。

2.串行和並行隊列的初始化函數。

3.異步分發調用

4.同步分發調用

接口聲明：

   @interface MADispatchQueue : NSObject
    
    + (MADispatchQueue *)globalQueue;
    
    - (id)initSerial: (BOOL)serial;
    
    - (void)dispatchAsync: (dispatch_block_t)block;
    - (void)dispatchSync: (dispatch_block_t)block;
    
    @end

接下來的目標就是實現這些方法的功能。

線程池接口

隊列後面的線程池接口更簡單。它將真正執行提交的任務。隊列負責在合適的時間把已入隊的任務提交給它。

線程池只做一件事：投遞任務並運行。對應地，一個接口只有一個方法：

@interface MAThreadPool : NSObject
    - (void)addBlock: (dispatch_block_t)block;
@end

由於這是核心部分，我們先實現它。

線程池實現

首先看實例變量。線程池能被多個內部線程或外部線程訪問，因此需要線程安全。而在可能的情況下，GCD會使用原子操作，而我這裡以一種以前比較流行的方式-加鎖。我需要知道鎖處於等待和鎖相關的信號，而不僅僅強制其互斥，因此我使用NSCondition而不是NSLock。如果你不熟悉，NSCondition 本質上還是鎖，只是添加了一個條件變量：

NSCondition *_lock;

想要知道什麼時候增加工作線程，我要知道線程池裡的線程數，有多少線程正被占用以及所能擁有的最大線程數：

    NSUInteger _threadCount;
    NSUInteger _activeThreadCount;
    NSUInteger _threadCountLimit;

最後，得有一個NSMutableArray類型的block列表模擬一個隊列，從隊列後端添加新block，從隊列前端刪除：

NSMutableArray *_blocks;

初始化函數很簡單。初始化鎖和block數組，隨便設置一個最大線程數比如128：

    - (id)init {
        if((self = [super init])) {
            _lock = [[NSCondition alloc] init];
            _blocks = [[NSMutableArray alloc] init];
            _threadCountLimit = 128;
        }
        return self;
    }

工作線程運行了一個簡單的無限循環。只要block數組為空，它將一直等待。一旦有block加入，它將被從數組中取出並執行。同時將活動線程數加1，完成後活動線程數減1：

- (void)workerThreadLoop: (id)ignore {

首先要獲取鎖。注意需要在循環開始前獲得。至於原因，等寫到循環結束時你就會明白。

        [_lock lock];

無限循環開始：

        while(1) {

如果隊列為空，等待鎖：

            while([_blocks count] == 0) {
                [_lock wait];
            }

注意：這裡是內循環結束而非if判斷。原因是由於虛假喚醒。簡單說來就是wait 在沒有信號通知的情況下也有可能返回，目前為此，條件檢測的正確方式是當wait 返回時重新進行條件檢測。

一旦有隊列中有block，取出：

            dispatch_block_t block = [_blocks firstObject];
            [_blocks removeObjectAtIndex: 0];

活動線程計數加，表示有新線程正在處理任務：

            _activeThreadCount++;

現在執行block，我們先得釋放鎖，不然代碼並發執行時會出現死鎖：

            [_lock unlock];

安全釋放鎖後，執行block

            block();

block執行完畢，活動線程計數減1。該操作必須在鎖內做，以避免競態條件，最後是循環結束：

            [_lock lock];
            _activeThreadCount--;
        }
    }

現在你該明白為什麼需要在進入循環前獲得鎖了。循環的最後是在鎖內減少活動線程計數。循環開始檢測block隊列。通過在循環外第一次獲得鎖，後續循環迭代能夠使用一個鎖來完成，而不是鎖，解鎖，然後再立即上鎖。

下面是 addBlock:

- (void)addBlock: (dispatch_block_t)block {

這裡唯一需要做的是獲得鎖：

        [_lock lock];

添加一個新的block到block隊列：

        [_blocks addObject: block];

如果有一個空閒的工作線程去執行這個block的話，這裡什麼都不需要做。如果沒有足夠的工作線程去處理等待的block，而工作線程數也沒超限，則我們需要創建一個新線程：

        NSUInteger idleThreads = _threadCount - _activeThreadCount;
        if([_blocks count] > idleThreads && _threadCount < _threadCountLimit) {
            [NSThread detachNewThreadSelector: @selector(workerThreadLoop:)
                                     toTarget: self
                                   withObject: nil];
            _threadCount++;
        }

一切准備就緒。由於空閒線程都在休眠，喚醒它：

        [_lock signal];

最後釋放鎖：

        [_lock unlock];
    }

線程池能在達到預設的最大線程數前創建工作線程，以處理對應的block。現在以此為基礎實現隊列。

隊列實現

和線程池一樣，隊列使用鎖保護其內容。和線程池不同的是，它不需要等待鎖，也不需要信號觸發，僅僅是簡單互斥即可，因此采用 NSLock：

    NSLock *_lock;

和線程池一樣，它把 pending block存在NSMutableArray裡。

NSMutableArray *_pendingBlocks;

標識是串行還是並行隊列：

BOOL _serial;

如果是串行隊列，還需要標識當前是否有線程正在運行：

BOOL _serialRunning;

並行隊列裡有無線程都一樣處理，所以無需關注。

全局隊列是一個全局變量，共享線程池也一樣。它們都在+initialize裡創建：

    static MADispatchQueue *gGlobalQueue;
    static MAThreadPool *gThreadPool;
    + (void)initialize {
        if(self == [MADispatchQueue class]) {
            gGlobalQueue = [[MADispatchQueue alloc] initSerial: NO];
            gThreadPool = [[MAThreadPool alloc] init];
        }
    }

由於+initialize裡已經初始化了，+globalQueue 只需返回該變量。

    + (MADispatchQueue *)globalQueue {
        return gGlobalQueue;
    }

這裡所做的事情和dispatch_once是一樣的，但是實現GCD API的時候使用GCD API有點自欺欺人，即使代碼不一樣。

初始化一個隊列：初始化lock 和pending Blocks，設置_serial變量：

    - (id)initSerial: (BOOL)serial {
        if ((self = [super init])) {
            _lock = [[NSLock alloc] init];
            _pendingBlocks = [[NSMutableArray alloc] init];
            _serial = serial;
        }
        return self;
    }

實現剩下的公有API前，我們需先實現一個底層方法用於給線程分發一個block，然後繼續調用自己去處理另一個block：

    - (void)dispatchOneBlock {

整個生命周期所做的是在線程池上運行block，分發代碼如下：

       [gThreadPool addBlock: ^{

然後取隊列中的第一個block，顯然這需要在鎖內完成，以避免出現問題：

            [_lock lock];
            dispatch_block_t block = [_pendingBlocks firstObject];
            [_pendingBlocks removeObjectAtIndex: 0];
            [_lock unlock];

取到了block又釋放了鎖，block接下來可以安全地在後台線程執行了：

            block();

如果是並行執行的話就不需要再做啥了。如果是串行執行，還需要以下操作：

            if(_serial) {

串行隊列裡將會積累別的block，但不能執行，直到先前的block完成。block完成後，dispatchOneBlock 接下來會看是否還有其他的block被添加到隊列裡面。若有，它調用自己去處理下一個block。若無，則把隊列的運行狀態置為NO:

                [_lock lock];
                if([_pendingBlocks count] > 0) {
                    [self dispatchOneBlock];
                } else {
                    _serialRunning = NO;
                }
                [_lock unlock];
            }
        }];
    }

用以上方法來實現dispatchAsync:就非常容易了。添加block到pending  block隊列，合適的時候設置狀態並調用dispatchOneBlock：

    - (void)dispatchAsync: (dispatch_block_t)block {
        [_lock lock];
        [_pendingBlocks addObject: block];

如果串行隊列空閒，設置隊列狀態為運行並調用dispatchOneBlock 進行處理。

        if(_serial && !_serialRunning) {
            _serialRunning = YES;
            [self dispatchOneBlock];

如果隊列是並行的，直接調用dispatchOneBlock。由於多個block能並行執行，所以這樣能保證即使有其他block正在運行，新的block也能立即執行。

        } else if (!_serial) {
            [self dispatchOneBlock];
        }

如果串行隊列已經在運行，則不需要另外做處理。因為block執行完成後對dispatchOneBlock 的調用最終會調用加入到隊列的block。接著釋放鎖：

        [_lock unlock];
    }

對於 dispatchSync: GCD的處理更巧妙，它是直接在調用線程上執行block，以防止其他block在隊列上執行（如果是串行隊列）。在此我們不用做如此聰明的處理，我們僅僅是對dispatchAsync:進行封裝，讓其一直等待直到block執行完成。

它使用局部NSCondition進行處理，另外使用一個done變量來指示block何時完成：

    - (void)dispatchSync: (dispatch_block_t)block {
        NSCondition *condition = [[NSCondition alloc] init];
        __block BOOL done = NO;

下面是異步分發block。block裡面調用傳入的block，然後設置done的值，給condition發信號

        [self dispatchAsync: ^{
            block();
            [condition lock];
            done = YES;
            [condition signal];
            [condition unlock];
        }];

在調用線程裡面，等待信號done ，然後返回

        [condition lock];
        while (!done) {
            [condition wait];
        }
        [condition unlock];
    }

到此。block的執行就結束了，這也是MADispatchQueue API的最後一點內容。

結論

全局線程池可以使用block隊列和智能產生的線程實現。使用一個共享全局線程池，就能構建一個能提供基本的串行/並行、同步/異步功能的dispatch queue。這樣就重建了一個簡單的GCD，雖然缺少了很多非常好的特性且更低效率。但這能讓我們瞥見其內部工作過程，揭示了它畢竟不是那麼神秘（除dispatch_once比較神秘外）

今天到此為止。下次再帶給大家有趣的東西，Friday Q&A內容取決於讀者的想法，因此如果你有什麼東西想在下次或以後了解的，請聯系我。
（譯者注：作者此前已經將網站上Friday Q&A系列文章整理成了一本書，開發者可在iBooks和Kindle上查看，另外還有PDF和ePub格式供下載。點擊此處查看詳細信息。)