iOS 底层探究：多线程之GCD分析（调度组与dispatch_source）

这是我参与8月更文挑战的第19天，活动详情查看：[8月更文挑战]

1. 调度组

调度组最直接的作用：控制任务执行顺序

• dispatch_group_creat：创建组
• dispatch_group_async：进组任务
• dispatch_group_notify：进组任务执行完毕通知
• dispatch_group_wait：进组任务执行等待时间
• dispatch_group_enter：进组
• dispatch_group_leave：出组

使用进组任务

- (void)testGCD{ 
    dispatch_group_t group = dispatch_group_create(); 
    dispatch_queue_t queue = dispatch_get_global_queue(0, 0); 
    dispatch_group_async(group, queue, ^{
        NSLog(@"接口1");
    }); 
    dispatch_group_async(group, queue, ^{ 
        NSLog(@"接口2"); 
    }); 
    dispatch_group_notify(group, dispatch_get_main_queue(), ^{ 
        NSLog(@"刷新");
    }); 
}
复制代码

使用进组、出组

- (void)testGCD{ 
    dispatch_group_t group = dispatch_group_create(); 
    dispatch_queue_t queue = dispatch_get_global_queue(0, 0);
    dispatch_group_enter(group); 
    dispatch_async(queue, ^{ 
        NSLog(@"接口1"); 
        dispatch_group_leave(group); 
    }); 
    dispatch_group_enter(group);
    dispatch_async(queue, ^{ 
        NSLog(@"接口2"); 
        dispatch_group_leave(group); 
    }); 
    dispatch_group_notify(group, dispatch_get_main_queue(), ^{ 
        NSLog(@"刷新");
    }); 
}
复制代码

• dispatch_group_enter和dispatch_group_leave必须成对使用，否则会一直等待或出现异常

1.1 创建组

进入dispatch_group_creat函数

dispatch_group_t dispatch_group_create(void) { 
    return _dispatch_group_create_with_count(0);
}
复制代码

进入_dispatch_group_creat_count函数

static inline dispatch_group_t _dispatch_group_create_with_count(uint32_t n) {
    dispatch_group_t dg = _dispatch_object_alloc(DISPATCH_VTABLE(group), sizeof(struct dispatch_group_s)); 
    dg->do_next = DISPATCH_OBJECT_LISTLESS;
    dg->do_targetq = _dispatch_get_default_queue(false); 
    if (n) { 
        os_atomic_store2o(dg, dg_bits, (uint32_t)-n * DISPATCH_GROUP_VALUE_INTERVAL, relaxed); 
        os_atomic_store2o(dg, do_ref_cnt, 1, relaxed); // <rdar://22318411> 
    } 
    return dg; 
}
复制代码

• 创建dispatch_group_t结构体，参数n默认传入0

1.2 进组

进入dispatch_group_enter函数

void dispatch_group_enter(dispatch_group_t dg) { 
    // The value is decremented on a 32bits wide atomic so that the carry 
    // for the 0 -> -1 transition is not propagated to the upper 32bits. 
    uint32_t old_bits = os_atomic_sub_orig2o(dg, dg_bits, DISPATCH_GROUP_VALUE_INTERVAL, acquire); 
    uint32_t old_value = old_bits & DISPATCH_GROUP_VALUE_MASK; 
    if (unlikely(old_value == 0)) { 
        _dispatch_retain(dg); // <rdar://problem/22318411> 
    }
    if (unlikely(old_value == DISPATCH_GROUP_VALUE_MAX)) {
        DISPATCH_CLIENT_CRASH(old_bits, "Too many nested calls to dispatch_group_enter()");
    }
}
复制代码

• 使用os_atomic_sub_orig2o宏，对dg_bits进行减1操作
• old_bits只可能是-1和0两种可能
• old_bits和DISPATCH_GROUP_VALUE_MASK进行&运算，将结果赋值给old_value
  • 如果old_bits为0，old_value为0，调用_dispatch_retain函数
  • 如果old_bits为-1，old_value为DISPATCH_GROUP_VALUE_MASK，标示进组和出组函数使用不平衡，报出异常

1.3 出组

进入dispatch_group_leave函数

void dispatch_group_leave(dispatch_group_t dg) { 
    // The value is incremented on a 64bits wide atomic so that the carry for 
    // the -1 -> 0 transition increments the generation atomically. 
    uint64_t new_state, old_state = os_atomic_add_orig2o(dg, dg_state, DISPATCH_GROUP_VALUE_INTERVAL, release); 
    uint32_t old_value = (uint32_t)(old_state & DISPATCH_GROUP_VALUE_MASK); 
    if (unlikely(old_value == DISPATCH_GROUP_VALUE_1)) { 
        old_state += DISPATCH_GROUP_VALUE_INTERVAL; 
        do { 
            new_state = old_state; 
            if ((old_state & DISPATCH_GROUP_VALUE_MASK) == 0) { 
                new_state &= ~DISPATCH_GROUP_HAS_WAITERS; 
                new_state &= ~DISPATCH_GROUP_HAS_NOTIFS; 
            } else { 
                // If the group was entered again since the atomic_add above, 
                // we can't clear the waiters bit anymore as we don't know for 
                // which generation the waiters are for 
                new_state &= ~DISPATCH_GROUP_HAS_NOTIFS; 
            }
            if (old_state == new_state) break; 
        } while (unlikely(!os_atomic_cmpxchgv2o(dg, dg_state, old_state, new_state, &old_state, relaxed))); 
        return _dispatch_group_wake(dg, old_state, true); 
    }
    if (unlikely(old_value == 0)) { 
        DISPATCH_CLIENT_CRASH((uintptr_t)old_value, "Unbalanced call to dispatch_group_leave()"); 
    }
}
复制代码

使用os_atomia_add_orig2o宏，进行加1操作

• &运算后的救治等于DISPATCH_GROUP_VALUE_1，等待do…while停止循环，调用_dispatch_group_wake函数
• DISPATCH_GROUP_VALUE_1等同于DISPATCH_GROUP_VALUE_MASK
• 如果旧值为0，表示进组和出组函数使用不平衡，爆出异常

进入_dispatch_group_wake函数

static void _dispatch_group_wake(dispatch_group_t dg, uint64_t dg_state, bool needs_release) { 
    uint16_t refs = needs_release ? 1 : 0; // <rdar://problem/22318411> 
    if (dg_state & DISPATCH_GROUP_HAS_NOTIFS) {
        dispatch_continuation_t dc, next_dc, tail;
        // Snapshot before anything is notified/woken <rdar://problem/8554546> 
        dc = os_mpsc_capture_snapshot(os_mpsc(dg, dg_notify), &tail); 
        do { 
            dispatch_queue_t dsn_queue = (dispatch_queue_t)dc->dc_data; 
            next_dc = os_mpsc_pop_snapshot_head(dc, tail, do_next); 
            _dispatch_continuation_async(dsn_queue, dc, _dispatch_qos_from_pp(dc->dc_priority), dc->dc_flags); 
            _dispatch_release(dsn_queue); 
        } while ((dc = next_dc)); 
        refs++; 
    }
    if (dg_state & DISPATCH_GROUP_HAS_WAITERS) {
        _dispatch_wake_by_address(&dg->dg_gen); 
    } 
    if (refs) _dispatch_release_n(dg, refs); 
}
复制代码

• 函数的作用，唤醒dispatch_group_notify函数
• 核心代码在do…while循环中，调用_dispatch_continuation_async函数

进入_dispatch_continuation_async函数

static inline void _dispatch_continuation_async(dispatch_queue_class_t dqu, dispatch_continuation_t dc, dispatch_qos_t qos, uintptr_t dc_flags) { 
#if DISPATCH_INTROSPECTION 
    if (!(dc_flags & DC_FLAG_NO_INTROSPECTION)) { 
        _dispatch_trace_item_push(dqu, dc); 
    }
#else 
    (void)dc_flags;
#endif 
    return dx_push(dqu._dq, dc, qos); 
}
复制代码

• 最终调用dx_push

1.4 通知

进入dispatch_group_notify函数

static inline void _dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq, dispatch_continuation_t dsn) 
{ 
    uint64_t old_state, new_state; 
    dispatch_continuation_t prev; 

    dsn->dc_data = dq; 
    _dispatch_retain(dq);

    prev = os_mpsc_push_update_tail(os_mpsc(dg, dg_notify), dsn, do_next);
    if (os_mpsc_push_was_empty(prev)) _dispatch_retain(dg);
    os_mpsc_push_update_prev(os_mpsc(dg, dg_notify), prev, dsn, do_next); 
    if (os_mpsc_push_was_empty(prev)) { 
        os_atomic_rmw_loop2o(dg, dg_state, old_state, new_state, release, { 
            new_state = old_state | DISPATCH_GROUP_HAS_NOTIFS; 
            if ((uint32_t)old_state == 0) { 
                os_atomic_rmw_loop_give_up({ 
                    return _dispatch_group_wake(dg, new_state, false);
                });
            }
        });
    }
}
复制代码

• 判断状态为0，调用_dispatch_group_wake函数
• 所以通知并不需要一直等待，因为dispatch_group_notify和dispatch_group_leave中都有_dispatch_group_wake函数的调用

1.5 任务

进入dispatch_group_async函数

void dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq, dispatch_block_t db) 
{
    dispatch_continuation_t dc = _dispatch_continuation_alloc(); 
    uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_GROUP_ASYNC; 
    dispatch_qos_t qos; 
    qos = _dispatch_continuation_init(dc, dq, db, 0, dc_flags); 
    _dispatch_continuation_group_async(dg, dq, dc, qos);
}
复制代码

进入_dispatch_continuation_group_async函数

static inline void _dispatch_continuation_group_async(dispatch_group_t dg, dispatch_queue_t dq, dispatch_continuation_t dc, dispatch_qos_t qos) 
{
    dispatch_group_enter(dg); 
    dc->dc_data = dg; 
    _dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
}
复制代码

• 调用dispatch_group_enter函数，进行进组操作

源码中搜索_dispatch_client_callout函数，在_dispatch_continuation_with_group_invoke函数中，同样调用dispatch_group_leave函数，进行出组操作

static inline void _dispatch_continuation_with_group_invoke(dispatch_continuation_t dc) 
{
    struct dispatch_object_s *dou = dc->dc_data;
    unsigned long type = dx_type(dou); 
    if (type == DISPATCH_GROUP_TYPE) { 
        _dispatch_client_callout(dc->dc_ctxt, dc->dc_func); 
        _dispatch_trace_item_complete(dc); 
        dispatch_group_leave((dispatch_group_t)dou); 
    } else { 
        DISPATCH_INTERNAL_CRASH(dx_type(dou), "Unexpected object type"); 
    }
}
复制代码

2. dispatch_source

dispatch_souce是基础数据类型，用于协调特定底层系统时间的处理

2.1基本介绍

dispatch_source替代了异步回调函数，来处理系统相关的事件。当配置一个dispatch时，你需要指定监测的事件、队列以及任务回调。当事件发生时，dispatch_source会提交block或函数到指定的queue区执行。
使用dispatch_source代替dispatch_async的原因在于联结的优势。
联结：在任一线程上调用它的一个函数dispatch_source_merge_data后，会执行Dispatch Source事先定义好的句柄（可以把句柄简单理解为一个block），这个过程叫Custom event，用户事件。是dispatch_source支持处理的一种事件。
句柄：是一种指向指针的指针，他指向的就是一个类或者结构，它和系统有密切的关系，包含：

• 实例句柄HINSTANCE
• 位图句柄HBITMAP
• 设备表句柄HDC
• 图标句柄HICON
• 通用句柄HANDLE

简单来说：这种事件是由你调用dispatch_source_merge_data函数来向自己发出的信号
使用dispatch_source的优点

• 其CPU符合非常小，尽量不占用资源
• 联结的优势

2.2 创建dispatch_source

使用dispatch_source_creat函数

dispatch_source_t source = dispatch_source_create(dispatch_source_type_t type, uintptr_t handle, unsigned long mask, dispatch_queue_t queue)
复制代码

• type：dispatch源可处理的事件
• handle：可以理解为句柄、索引或ID，假如要监听进程，需要传入进程ID
• mask：可以理解为描述，提供更详细的描述，让它知道具体要监听什么
• queue：自定义源需要的一个队列，用来处理所有的响应句柄

dispatch_source中type的类型

• ：自定义的事件，变量增加
• DISPATCH_SOURCE_TYPE_DATA_OR：自定义的事件，变量OR
• DISPATCH_SOURCE_TYPE_MACH_SENDMACH：端口发送
• DISPATCH_SOURCE_TYPE_MACH_RECVMACH：端口接收
• DISPATCH_SOURCE_TYPE_MEMORYPRESSURE：内存压力 (注：iOS8后可用)
• DISPATCH_SOURCE_TYPE_PROC：进程监听,如进程的退出、创建一个或更多的子线程、进程收到UNIX信号
• DISPATCH_SOURCE_TYPE_READ：IO操作，如对文件的操作、socket操作的读响应
• DISPATCH_SOURCE_TYPE_SIGNAL：接收到UNIX信号时响应
• DISPATCH_SOURCE_TYPE_TIMER：定时器
• DISPATCH_SOURCE_TYPE_VNODE：文件状态监听，文件被删除、移动、重命名
• DISPATCH_SOURCE_TYPE_WRITE：IO操作，如对文件的操作、socket操作的写响应

2.3 常用API

//挂起队列
dispatch_suspend(queue) 

//分派源创建时默认处于暂停状态，在分派源分派处理程序之前必须先恢复 
dispatch_resume(source) 

//向分派源发送事件，需要注意的是，不可以传递0值(事件不会被触发)，同样也不可以传递负数。 
dispatch_source_merge_data 

//设置响应分派源事件的block，在分派源指定的队列上运行 
dispatch_source_set_event_handler 

//得到分派源的数据 
dispatch_source_get_data
 
//得到dispatch源创建，即调用dispatch_source_create的第二个参数 
uintptr_t dispatch_source_get_handle(dispatch_source_t source); 

//得到dispatch源创建，即调用dispatch_source_create的第三个参数 
unsigned long dispatch_source_get_mask(dispatch_source_t source);

////取消dispatch源的事件处理--即不再调用block。如果调用dispatch_suspend只是暂停dispatch源。 
void dispatch_source_cancel(dispatch_source_t source); 

//检测是否dispatch源被取消，如果返回非0值则表明dispatch源已经被取消 
long dispatch_source_testcancel(dispatch_source_t source);

//dispatch源取消时调用的block，一般用于关闭文件或socket等，释放相关资源 
void dispatch_source_set_cancel_handler(dispatch_source_t source, dispatch_block_t cancel_handler); 

//可用于设置dispatch源启动时调用block，调用完成后即释放这个block。也可在dispatch源运行当中随时调用这个函数。
void dispatch_source_set_registration_handler(dispatch_source_t source, dispatch_block_t registration_handler);
复制代码

2.4 timer的封装

打开SparkTimer.h文件，写入以下代码：

#import <Foundation/Foundation.h> 
@class SparkTimer; 
typedef void (^TimerBlock)(SparkTimer * _Nonnull timer); 
@interface SparkTimer : NSObject
+ (SparkTimer *)scheduledTimerWithTimeInterval:(NSTimeInterval)ti target:(id)aTarget selector:(SEL)aSelector userInfo:(nullable id)userInfo repeats:(BOOL)yesOrNo immediately:(BOOL)isImmediately; 
+ (SparkTimer *)scheduledTimerWithTimeInterval:(NSTimeInterval)ti userInfo:(nullable id)userInfo repeats:(BOOL)yesOrNo immediately:(BOOL)isImmediately timerBlock:(TimerBlock)block; 
- (void)invalidate; 
@property (nonatomic, readonly, getter=isValid) BOOL valid; 
@property (nonatomic, nullable, readonly, retain) id userInfo;
@end
复制代码

打开SparkTimer.m文件，写入以下代码：

#import "SparkTimer.h" 
@interface SparkTimer ()
@property (nonatomic, assign) NSTimeInterval interval;
@property (nonatomic, nullable, readwrite, retain) id userInfo; 
@property (nonatomic, assign) BOOL repeats;
@property (nonatomic, assign) BOOL immediately; 
@property (nonatomic, strong) dispatch_source_t timer;
@property (nonatomic, readwrite, getter=isValid) BOOL valid;
@end 

@implementation SparkTimer 

+ (SparkTimer *)scheduledTimerWithTimeInterval:(NSTimeInterval)ti target:(id)aTarget selector:(SEL)aSelector userInfo:(nullable id)userInfo repeats:(BOOL)yesOrNo immediately:(BOOL)isImmediately{ 
    return [[SparkTimer alloc] initWithInterval:ti target:aTarget selector:aSelector userInfo:userInfo repeats:yesOrNo immediately:isImmediately timerBlock:nil isBlock:NO]; 
} 
+ (SparkTimer *)scheduledTimerWithTimeInterval:(NSTimeInterval)ti userInfo:(nullable id)userInfo repeats:(BOOL)yesOrNo immediately:(BOOL)isImmediately timerBlock:(TimerBlock)block{ 
    return [[SparkTimer alloc] initWithInterval:ti target:nil selector:nil userInfo:userInfo repeats:yesOrNo immediately:isImmediately timerBlock:block isBlock:YES];
}

- (instancetype)initWithInterval:(NSTimeInterval)ti target:(id)aTarget selector:(SEL)aSelector userInfo:(nullable id)userInfo repeats:(BOOL)yesOrNo immediately:(BOOL)isImmediately timerBlock:(TimerBlock)block isBlock:(BOOL)isBlock { 
    self = [super init]; 
    if (self) {
        _interval = ti; 
        _userInfo = userInfo;
        _repeats = yesOrNo;
        _immediately = isImmediately; 
        _valid = NO; 
        @weakify(self) 
        [self createTimer:^{ 
            @strongify(self) 
            if(!isBlock){ 
                [self callOutWithTarget:aTarget selector:aSelector]; 
                return; 
            } 
            [self callOutWithTimerBlock:block]; 
        }]; 
    } 
    return self; 
}
- (void)callOutWithTarget:(id)aTarget selector:(SEL)aSelector{
    if(!aTarget || !aSelector){ 
        [self invalidate]; 
        return;
    } 
    [aTarget performSelector:aSelector withObject:self]; 
    [self checkRepeats]; 
} 
- (void)callOutWithTimerBlock:(TimerBlock)block{ 
    if(!block){ 
        [self invalidate]; 
        return; 
    } 
    block(self);
    [self checkRepeats]; 
}
- (void)checkRepeats{ 
    if(self.repeats){ 
        return;
    }
    [self invalidate]; 
}
- (void)createTimer:(void(^)(void))block{
    _valid = YES;
    
    //1.创建队列
    dispatch_queue_t queue = dispatch_get_global_queue(0, 0);
    //2.创建timer 
    _timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, queue); 
    dispatch_time_t start; 
    if(self.immediately){
        start = DISPATCH_TIME_NOW; 
    } else{ 
        start = dispatch_time(DISPATCH_TIME_NOW, self.interval * NSEC_PER_SEC);
    } 
    //3.设置timer首次执行时间，间隔，精确度 
    dispatch_source_set_timer(_timer, start, self.interval * NSEC_PER_SEC, 0.1 * NSEC_PER_SEC);
    //4.设置timer事件回调 
    dispatch_source_set_event_handler(_timer, ^{ 
        NSLog(@"计时"); 
        block();
    }); 
    //5.默认是挂起状态，需要手动激活 
    dispatch_resume(_timer); 
}
- (void)invalidate{ 
    if(!self.isValid){ 
        return; 
    }
    _valid = NO;
    dispatch_source_cancel(_timer); 
}
- (id)userInfo{ 
    return _userInfo;
}
@end
复制代码