IT博客汇
  • 首页
  • 精华
  • 技术
  • 设计
  • 资讯
  • 扯淡
  • 权利声明
    • 登录 注册

    并发编程之GCD

    Xcodev发表于 2016-03-11 02:33:57
    love 0

    在《并发编程之Operation Queue》中讲了Cocoa并发编程中的Operation Queue，了解了Operation Queue是一个面向对象的并发编程接口，它支持并发数，线程优先级，任务优先级，任务依赖关系等多种配置，可以方便满足各种复杂的多任务处理场景。本篇将接着讲另一种并发编程机制 - GCD（Grand Central Dispatch）。iOS4.0中首度引入GCD，GCD是管理任务执行的一项技术，它使得我们对多任务处理变得更加方便和有效。它支持同步或异步任务处理，串行或并行的处理队列（Dispath Queue），非系统调用的信号量机制，定时任务处理，进程、文件或网络的监听任务等。这个庞大的任务处理技术大大减少了线程的管理工作，使基于任务的开发变得更加高效。

    Dispatch Queue

    Dispatch Queue是一个任务执行队列，可以让你异步或同步地执行多个Block或函数。Dispatch Queue是FIFO的，即先入队的任务总会先执行。目前有三种类型的Dispath Queue：

    • 串行队列（Serial dispatch queue）
    • 并发队列（Concurrent dispatch queue）
    • 主队列（Main dispatch queue）

    串行队列

    串行队列一次只能处理一个任务，可以由用户调用dispatch_queue_create创建：

    1
    2
    dispatch_queue_t queue;
    queue = dispatch_queue_create("com.example.MyQueue", NULL);

    dispatch_queue_create第一个参数是串行队列标识，一般用反转域名的格式表示以防冲突；第二个参数是queue的类型，设为NULL时默认是DISPATCH_QUEUE_SERIAL，将创建串行队列，在必要情况下，你可以将其设置为DISPATCH_QUEUE_CONCURRENT来创建自定义并行队列。

    并行队列

    并行队列可以同时处理多个任务，在不得以的情况下可以用dispatch_queue_create创建，但一般我们都要用系统预定义的并行队列，即全局队列（Global Concurrent Dispatch Queues）。目前系统预定义了四个不同运行优先级的全局队列，我们可以通过dispatch_get_global_queue来获取它们。

    1
    dispatch_queue_t aQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);

    dispatch_get_global_queue第一个参数是队列的优先级，分别对应四个全局队列：

    • DISPATCH_QUEUE_PRIORITY_HIGH
    • DISPATCH_QUEUE_PRIORITY_DEFAULT
    • DISPATCH_QUEUE_PRIORITY_LOW
    • DISPATCH_QUEUE_PRIORITY_BACKGROUND

    dispatch_get_global_queue中第二个参数目前系统保留，请设置为0即可。

    主队列

    主队列是一个特殊的队列，它是系统预定义的运行在主线程的一个Dispatch Queue。可以通过dispatch_get_main_queue来获取唯一的主队列。主队列一般运行一些需要与主线程同步的一些短时任务。

    1
    dispatch_queue_t mainQueue = dispatch_get_main_queue();

    获取当前队列

    你可以通过dispatch_get_current_queue获取运行时的队列：

    1
    dispatch_queue_t currentQueue = dispatch_get_current_queue();

    如果在队列执行任务中调用，返回执行此任务的队列；如果在主线程中调用，将返回主队列；如果在一般线程（非主线程线程非队列执行任务）中调用，返回DISPATCH_QUEUE_PRIORITY_DEFAULT全局队列。

    在队列中运行任务

    你可以随时向一个队列中添加一个新任务，只需要调用一下dispatch_async即可：

    1
    2
    3
    dispatch_async(aQueue, ^{
        //Do some work;
    });

    dispatch_async中的任务是异步执行的，就是说dispatch_async添加任务到执行队列后会立刻返回，而不会等待任务执行完成。然而，必要的话，你也可以调用dispatch_sync来同步的执行一个任务：

    1
    2
    3
    dispatch_sync(aQueue, ^{
        //Do some work;
    });

    dispatch_sync会阻塞当前线程直到提交的任务完全执行完毕。

    Dispatch Queue的内存管理

    除了系统预定义的Dispatch Queue，我们自定义的Dispatch Queue需要手动的管理它的内存。dispatch_retain和dispatch_release这两个函数可以控制Dispatch Queue的引用计数（同时可以控制后面会讲到的Dispatch Group和Dispatch Source的引用计数）。当Dispatch Queue引用计数变为0后，就会调用finalizer，finalizer是Dispatch Queue销毁前调用的函数，用来清理Dispatch Queue的相关资源。可以用dispatch_set_finalizer_f函数来设置Dispatch Queue的finalizer，这个函数同时可以设置Dispatch Group和Dispatch Source的销毁函数（后面会讲到）。

    1
    void dispatch_set_finalizer_f(dispatch_object_t object, dispatch_function_t finalizer);

    Dispatch Queue的上下文环境数据

    我们可以为每个Dispatch Queue设置一个自定义的上下文环境数据，调用dispatch_set_context来实现。同时我们也可以用dispatch_get_context获取这个上下文环境数据，这个函数同时可以设置Dispatch Group和Dispatch Source的上下文环境数据（后面会讲到）。

    1
    2
    void dispatch_set_context(dispatch_object_t object,void *context);
    void * dispatch_get_context(dispatch_object_t object);

    注意Dispatch Queue并不保证这个context不会释放，不会对它进行内存管理控制。我们需要自行管理context的内存分配和释放。一般我们非配内存设置context后，可以在finalizer里释放context占有的内存。

    并行执行循环

    在编程过程中，我们经常会用到for循环，而且for循环要做很多相关的任务。比如：

    1
    2
    3
    4
    for (i = 0; i < count; i++) {
       //do a lot of work here.
       doSomething(i);
    }

    如果for循环中处理的任务是可并发的，显然放到一个线程中处理是很慢的，GCD提供两个函数dispatch_apply和dispatch_apply_f，dispatch_apply是用于Block的，而dispatch_apply_f可以用于c函数，它们可以替代可并发的for循环，来并行的运行而提高执行效率。

    1
    2
    3
    4
    5
    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
    dispatch_apply(count, queue, ^(size_t i) {
       //do a lot of work here.
       doSomething(i);
    });

    Dispatch Group

    有时候我们进行下一步操作，而这个操作需要等待几个任务处理完毕后才能继续，这时我们就需要用的Dispatch Group（类似thread join）。我们可以把若干个任务放到一个Dispatch Group中：

    1
    2
    3
    4
    5
    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
    dispatch_group_t group = dispatch_group_create();
    dispatch_group_async(group, queue, ^{
       // Some asynchronous work
    });

    dispatch_group_async跟dispatch_async一样，会把任务放到queue中执行，不过它比dispatch_async多做了一步操作就是把这个任务和group相关联。

    把一些任务放到Dispatch Group后，我们就可以调用dispatch_group_wait来等待这些任务完成。若任务已经全部完成或为空，则直接返回，否则等待所有任务完成后返回。注意：返回后group会清空。

    1
    2
    3
    dispatch_group_wait(group, DISPATCH_TIME_FOREVER);
    // Do some work after.
    dispatch_release(group);

    Dispatch信号量

    很多程序设计都设计到信号量，生产者-消费者模型在多线程编程中会频繁的使用。GCD提供了自己的一套信号量机制。

    信号量就是一些有限可数资源。比如打印机，假如系统有2台打印机，但同时有5个任务要使用打印机，那么只能有2个任务能同时进行打印，剩下3个要等待这2个任务打印完。那么程序工作过程应该是：任务首先获取打印机资源（dispatch_semaphore_wait），如果没有打印机可用了就要等待，直到其他任务用完这个打印机。当任务获取到打印机，就开始执行打印任务。任务用完打印机工作后，就必须把占用打印机释放（dispatch_semaphore_signal），以便其他任务可以接着打印。

    1
    2
    3
    4
    5
    6
    7
    8
    //创建资源的信号量，只创建一次，比如2台打印机，那么RESOURCE_SIZE为2.
    dispatch_semaphore_t sema = dispatch_semaphore_create(RESOURCE_SIZE);
    //如果任务使用一个资源时，使用前调用dispatch_semaphore_wait，用完后dispatch_semaphore_signal。
    //下面的代码可能在多个线程中调用多次
    dispatch_semaphore_wait(sema, DISPATCH_TIME_FOREVER);
    //do some work here. the work will use one resource.
    dispatch_semaphore_signal(sema);

    信号量必须在资源使用之前调用dispatch_semaphore_create创建，而且只创建一次，RESOURCE_SIZE是可用资源的总数。使用资源时dispatch_semaphore_wait将资源的可用数减少一个，如果当前没有可用资源了，将会等待直到其他线程回收资源，即调用dispatch_semaphore_signal让可用资源增加。用完资源后调用dispatch_semaphore_signal回收可利用资源，资源可用数将增加一个。

    如果是生产者-消费者模型的话，RESOURCE_SIZE可能最初为0，那么生产者将调用dispatch_semaphore_signal来产生一个单位的资源，消费者调用dispatch_semaphore_wait来消费（减少）一个单位的资源。当资源不足时，消费者会一直等待到资源数大于0，即生产者生成新的资源。

    Dispatch Source

    Dispatch Source是GCD中监听一些系统事件的有个Dispatch对象，它包括定时器、文件监听、进程监听、Mach port监听等类型。

    可以通过dispatch_source_create创建一个Dispatch Source：

    1
    2
    3
    4
    5
    dispatch_source_t dispatch_source_create(
       dispatch_source_type_t type,
       uintptr_t handle,
       unsigned long mask,
       dispatch_queue_t queue);

    这里可以指定Dispatch Source的类型，type可以为文件读或写、进程监听等。handle为监听对象的句柄，如果是文件就是文件描述符，如果是进程就是进程ID。mask用来指定一些想要监听的事件，它的意义取决于type。queue指定事件处理的任务队列。

    创建好Dispatch Source后，我们要为Dispatch Source设置一个事件处理模块。可以用dispatch_source_set_event_handler或dispatch_source_set_event_handler_f来设置：

    1
    2
    3
    void dispatch_source_set_event_handler(
       dispatch_source_t source,
       dispatch_block_t handler);

    设置好Dispatch Source后就可以调用dispatch_resume来启动监听。如果相应的事件发生就会触发事件处理模块。

    同时我们也可以设置一个取消处理模块：

    1
    2
    3
    dispatch_source_set_cancel_handler(mySource, ^{
       close(fd); // Close a file descriptor opened earlier.
    });

    取消处理模块会在Dispatch Source取消时调用。

    下面介绍一下主要的Dispatch Source类型和示例代码。

    定时器

    定时器Dispatch Source可以每隔一个固定的时间处理一下任务。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    dispatch_source_t CreateDispatchTimer(uint64_t interval,
                  uint64_t leeway,
                  dispatch_queue_t queue,
                  dispatch_block_t block)
    {
       dispatch_source_t timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER,
                                                         0, 0, queue);
       if (timer)
       {
          dispatch_source_set_timer(timer, dispatch_walltime(NULL, 0), interval, leeway);
          dispatch_source_set_event_handler(timer, block);
          dispatch_resume(timer);
       }
       return timer;
    }
     
    void MyCreateTimer()
    {
       dispatch_source_t aTimer = CreateDispatchTimer(30ull * NSEC_PER_SEC,
                                   1ull * NSEC_PER_SEC,
                                   dispatch_get_main_queue(),
                                   ^{ MyPeriodicTask(); });
     
       // Store it somewhere for later use.
        if (aTimer)
        {
            MyStoreTimer(aTimer);
        }
    }

    dispatch_after和dispatch_after_f

    有时候我们只想处理一次延迟任务，可以用dispatch_after和dispatch_after_f

    1
    2
    3
    4
    void dispatch_after(
       dispatch_time_t when,
       dispatch_queue_t queue,
       dispatch_block_t block);

    监听文件事件

    监听文件事件分好几个类型，有读、写、属性的监听。

    读取文件

    1
    2
    3
    4
    5
    6
    7
    8
    dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, fd, 0, queue);
    dispatch_source_set_event_handler(source, ^{
       // Get some data from the source variable, which is captured
       // from the parent context.
       size_t estimated = dispatch_source_get_data(source);
       // Continue reading the descriptor...
    });
    dispatch_resume(source);

    写文件

    1
    2
    3
    4
    5
    6
    7
    dispatch_source_t writeSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE,
                            fd, 0, queue);
    if (!writeSource)
    {
        close(fd);
        return NULL;
    }
    1
    2
    3
    dispatch_source_set_event_handler(writeSource, ^{
        size_t bufferSize = MyGetDataSize();
        void* buffer = malloc(bufferSize);
    1
    2
    size_t actual = MyGetData(buffer, bufferSize);
    write(fd, buffer, actual);
    1
    free(buffer);
    1
    2
    3
        // Cancel and release the dispatch source when done.
        dispatch_source_cancel(writeSource);
    });

    监听文件属性

    1
    2
    3
    4
    5
    6
    7
    8
    9
    dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_VNODE,
                fd, DISPATCH_VNODE_RENAME, queue);
    if (source)
    {
      // Copy the filename for later use.
      int length = strlen(filename);
      char* newString = (char*)malloc(length + 1);
      newString = strcpy(newString, filename);
      dispatch_set_context(source, newString);
    1
    2
    3
    4
    5
    // Install the event handler to process the name change
    dispatch_source_set_event_handler(source, ^{
          const char*  oldFilename = (char*)dispatch_get_context(source);
          MyUpdateFileName(oldFilename, fd);
    });
    1
    2
    3
    4
    5
    6
    7
    // Install a cancellation handler to free the descriptor
    // and the stored string.
    dispatch_source_set_cancel_handler(source, ^{
        char* fileStr = (char*)dispatch_get_context(source);
        free(fileStr);
        close(fd);
    });
    1
    2
    3
    4
    5
      // Start processing events.
      dispatch_resume(source);
    }
    else
      close(fd);

    监听进程事件

    DISPATCH_PROC_EXIT是一个监听进程退出的类型。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_PROC,
                                                      parentPID, DISPATCH_PROC_EXIT, queue);
    if (source)
    {
       dispatch_source_set_event_handler(source, ^{
         MySetAppExitFlag();
         dispatch_source_cancel(source);
         dispatch_release(source);
       });
       dispatch_resume(source);
    }

    监听中断信号

    1
    2
    3
    4
    5
    6
    dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_SIGNAL, SIGHUP, 0, queue);
    if (source)
    {
      dispatch_source_set_event_handler(source, ^{
         MyProcessSIGHUP();
      });
    1
    2
    3
      // Start processing signals
      dispatch_resume(source);
    }

    参考文献

    • Dispatch Queues：https://developer.apple.com/library/mac/documentation/general/conceptual/concurrencyprogrammingguide/OperationQueues/OperationQueues.html#//apple_ref/doc/uid/TP40008091-CH102-SW1
    • Dispatch Sources：https://developer.apple.com/library/mac/documentation/general/conceptual/concurrencyprogrammingguide/GCDWorkQueues/GCDWorkQueues.html


沪ICP备19023445号-2号
友情链接