深入理解dispatch_queue

文本由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 比较神秘外）