Laucp's Blog
0%

ReactiveCocoa-RACScheduler底层实现分析

发表于 更新于 分类于

RACScheduler ReactiveCocoa 框架中的调度器，ReactiveCocoa 中的信号可以在 RACScheduler 上执行任务、发送结果； RACScheduler 的实现主要是基于 GCD 封装，提供了 GCD 不具备的特性。

RACScheduler 有 4 个子类:

  • RACTestScheduler

  • RACSubscriptionScheduler

  • RACImmediateScheduler

  • RACQueueScheduler

接下来会分析平时使用中常接触到的 RACImmediateScheduler、RACQueueScheduler 和 RACSubscriptionScheduler 3个子类

RACScheduler

基类提供了5个构造方法：

1
2
3
4
5
6
7
8
9
10
11
+ (RACScheduler *)schedulerWithPriority:(RACSchedulerPriority)priority name:(nullable NSString *)name;

+ (RACScheduler *)schedulerWithPriority:(RACSchedulerPriority)priority;

+ (RACScheduler *)scheduler;

+ (nullable RACScheduler *)currentScheduler;

+ (RACScheduler *)immediateScheduler;

+ (RACScheduler *)mainThreadScheduler;

+schedulerWithPriority:name:、+schedulerWithPriority:、+scheduler 三个方法细线逻辑相差不大

1
2
3
4
5
6
7
8
9
10
11
+ (RACScheduler *)schedulerWithPriority:(RACSchedulerPriority)priority name:(NSString *)name {
	return [[RACTargetQueueScheduler alloc] initWithName:name targetQueue:dispatch_get_global_queue(priority, 0)];
}

+ (RACScheduler *)schedulerWithPriority:(RACSchedulerPriority)priority {
	return [self schedulerWithPriority:priority name:@"org.reactivecocoa.ReactiveObjC.RACScheduler.backgroundScheduler"];
}

+ (RACScheduler *)scheduler {
	return [self schedulerWithPriority:RACSchedulerPriorityDefault];
}
  1. +schedulerWithPriority:name:: 指定线程的优先级和名称，返回的 RACTargetQueueScheduler 对象，RACTargetQueueScheduler 是 RACQueueScheduler 的子类
  2. +schedulerWithPriority:: 指定线程的优先级，线程名称设置为 org.reactivecocoa.ReactiveObjC.RACScheduler.backgroundScheduler
  3. +scheduler: 线程的优先级设置为 RACSchedulerPriorityDefault，线程名称设置为 org.reactivecocoa.ReactiveObjC.RACScheduler.backgroundScheduler

+currentScheduler：

1
2
3
4
5
6
7
8
9
10
11
+ (RACScheduler *)currentScheduler {
	RACScheduler *scheduler = NSThread.currentThread.threadDictionary[RACSchedulerCurrentSchedulerKey];
	if (scheduler != nil) return scheduler;
	if ([self.class isOnMainThread]) return RACScheduler.mainThreadScheduler;

	return nil;
}

+ (BOOL)isOnMainThread {
	return [NSOperationQueue.currentQueue isEqual:NSOperationQueue.mainQueue] || [NSThread isMainThread];
}

该方法是从当前线程的线程字典获取对应的 RACScheduler 对象，如果为 nil 则判断当前线程是否为主线程，如果是在主线程上，就返回 mainThreadScheduler。如果既不在主线程上，则返回 nil。

+immediateScheduler:

1
2
3
4
5
6
7
8
9
+ (RACScheduler *)immediateScheduler {
	static dispatch_once_t onceToken;
	static RACScheduler *immediateScheduler;
	dispatch_once(&onceToken, ^{
		immediateScheduler = [[RACImmediateScheduler alloc] init];
	});
	
	return immediateScheduler;
}

这是一个单例方法，返回了 RACImmediateScheduler 对象

+mainThreadScheduler

1
2
3
4
5
6
7
8
9
+ (RACScheduler *)mainThreadScheduler {
	static dispatch_once_t onceToken;
	static RACScheduler *mainThreadScheduler;
	dispatch_once(&onceToken, ^{
		mainThreadScheduler = [[RACTargetQueueScheduler alloc] initWithName:@"org.reactivecocoa.ReactiveObjC.RACScheduler.mainThreadScheduler" targetQueue:dispatch_get_main_queue()];
	});
	
	return mainThreadScheduler;
}

同样是一个单例方法，返回的是 RACTargetQueueScheduler 对象，并将其名称赋值为 org.reactivecocoa.ReactiveObjC.RACScheduler.mainThreadScheduler

+subscriptionScheduler

1
2
3
4
5
6
7
8
9
+ (RACScheduler *)subscriptionScheduler {
	static dispatch_once_t onceToken;
	static RACScheduler *subscriptionScheduler;
	dispatch_once(&onceToken, ^{
		subscriptionScheduler = [[RACSubscriptionScheduler alloc] init];
	});

	return subscriptionScheduler;
}

也是一个单例方法，返回了 RACSubscriptionScheduler 对象。

RACScheduler 还提供了5个实例方法：

1
2
3
4
5
6
7
8
9
- (nullable RACDisposable *)schedule:(void (^)(void))block;

- (nullable RACDisposable *)after:(NSDate *)date schedule:(void (^)(void))block;

- (nullable RACDisposable *)afterDelay:(NSTimeInterval)delay schedule:(void (^)(void))block;

- (nullable RACDisposable *)after:(NSDate *)date repeatingEvery:(NSTimeInterval)interval withLeeway:(NSTimeInterval)leeway schedule:(void (^)(void))block;

- (nullable RACDisposable *)scheduleRecursiveBlock:(RACSchedulerRecursiveBlock)recursiveBlock;

  1. schedule:: 传入 block 并根据相关条件来触发 block

  2. after:schedule:: 延迟执行 block，传入的延迟时间是 NSDate 类型

  3. afterDelay:schedule:: 延迟执行 block，传入的延迟时间是 NSTimeInterval 类型

  4. after:repeatingEvery:withLeeway:schedule:: 创建定时任务，循环间隔由参数 interval 决定

  5. scheduleRecursiveBlock:: 递归触发 block

上面5个方法除了 -scheduleRecursiveBlock:addingToDisposable: 都是由子类进行具体实现

-scheduleRecursiveBlock:addingToDisposable:

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
- (void)scheduleRecursiveBlock:(RACSchedulerRecursiveBlock)recursiveBlock addingToDisposable:(RACCompoundDisposable *)disposable {
	@autoreleasepool {
		RACCompoundDisposable *selfDisposable = [RACCompoundDisposable compoundDisposable];
		[disposable addDisposable:selfDisposable];

		__weak RACDisposable *weakSelfDisposable = selfDisposable;

		RACDisposable *schedulingDisposable = [self schedule:^{
			@autoreleasepool {
        // 已经触发之后，weakSelfDisposable 已经没有作用，故移除
				[disposable removeDisposable:weakSelfDisposable];
			}

			if (disposable.disposed) return;

			void (^reallyReschedule)(void) = ^{
				if (disposable.disposed) return;
				[self scheduleRecursiveBlock:recursiveBlock addingToDisposable:disposable];
			};

			// Protects the variables below.
			//
			// This doesn't actually need to be __block qualified, but Clang
			// complains otherwise. :C
			__block NSLock *lock = [[NSLock alloc] init];
			lock.name = [NSString stringWithFormat:@"%@ %s", self, sel_getName(_cmd)];

			__block NSUInteger rescheduleCount = 0;

      // 同步操作执行完后会赋值为YES，然后执行reallyReschedule闭包
			__block BOOL rescheduleImmediately = NO;

			@autoreleasepool {
				recursiveBlock(^{
					[lock lock];
					BOOL immediate = rescheduleImmediately;
					if (!immediate) ++rescheduleCount;
					[lock unlock];

					if (immediate) reallyReschedule();
				});
			}

			[lock lock];
			NSUInteger synchronousCount = rescheduleCount;
			rescheduleImmediately = YES;
			[lock unlock];

			for (NSUInteger i = 0; i < synchronousCount; i++) {
				reallyReschedule();
			}
		}];

		[selfDisposable addDisposable:schedulingDisposable];
	}
}

上面这段函数实现中有几个关键的变量/参数：

  • reallyReschedule：递归执行函数的闭包

  • recursiveBlock：函数传参，该闭包的参数也是一个闭包(block)，recursiveBlock 执行完之后会触发传入的闭包 block

  • rescheduleCount：递归的次数

  • rescheduleImmediately：是否立即执行递归闭包 reallyReschedule

主要流程:

  1. 初始化相关变量之后，执行 recursiveBlock ，第一次 rescheduleImmediately 为 NO，rescheduleCount 递增
  2. recursiveBlock 执行完后 rescheduleImmediately 赋值为 YES
  3. 递归执行 reallyReschedule 闭包 rescheduleCount 次
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
- (void)performAsCurrentScheduler:(void (^)(void))block {
	NSCParameterAssert(block != NULL);

	// If we're using a concurrent queue, we could end up in here concurrently,
	// in which case we *don't* want to clear the current scheduler immediately
	// after our block is done executing, but only *after* all our concurrent
	// invocations are done.

	RACScheduler *previousScheduler = RACScheduler.currentScheduler;
	NSThread.currentThread.threadDictionary[RACSchedulerCurrentSchedulerKey] = self;

	@autoreleasepool {
		block();
	}

	if (previousScheduler != nil) {
		NSThread.currentThread.threadDictionary[RACSchedulerCurrentSchedulerKey] = previousScheduler;
	} else {
		[NSThread.currentThread.threadDictionary removeObjectForKey:RACSchedulerCurrentSchedulerKey];
	}
}
  1. 执行 block 之前将当前线程的 scheduler 保存下来为 previousScheduler，然后设置自己为新的 scheduler
  2. 执行完 block 之后恢复现场，如果 previousScheduler 不为空，则重新复制为当前线程的 scheduler，否则从 threadDictionary 移除 RACSchedulerCurrentSchedulerKey 对应的对象，也就是当前 scheduler

RACImmediateScheduler

RACImmediateScheduler 是一个私有类，主要特点是将加入的 block 立即(Immediate)进行调度

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
30
31
- (RACDisposable *)schedule:(void (^)(void))block {
	NSCParameterAssert(block != NULL);

	block();
	return nil;
}

- (RACDisposable *)after:(NSDate *)date schedule:(void (^)(void))block {
	NSCParameterAssert(date != nil);
	NSCParameterAssert(block != NULL);

	[NSThread sleepUntilDate:date];
	block();

	return nil;
}

- (RACDisposable *)after:(NSDate *)date repeatingEvery:(NSTimeInterval)interval withLeeway:(NSTimeInterval)leeway schedule:(void (^)(void))block {
	NSCAssert(NO, @"+[RACScheduler immediateScheduler] does not support %@.", NSStringFromSelector(_cmd));
	return nil;
}

- (RACDisposable *)scheduleRecursiveBlock:(RACSchedulerRecursiveBlock)recursiveBlock {
	for (__block NSUInteger remaining = 1; remaining > 0; remaining--) {
		recursiveBlock(^{
			remaining++;
		});
	}

	return nil;
}
  • -schedule 方法会立即触发传入的 block；
  • -after:schedule: 方法会将当前线程休眠到指定时间后执行 block；
  • -after:repeatingEvery:withLeeway:schedule: 不支持
  • -scheduleRecursiveBlock: 循环不断执行传入的 block

RACQueueScheduler

1
2
3
4
5
6
7
8
9
10
11
12
- (RACDisposable *)schedule:(void (^)(void))block {
	NSCParameterAssert(block != NULL);

	RACDisposable *disposable = [[RACDisposable alloc] init];

	dispatch_async(self.queue, ^{
		if (disposable.disposed) return;
		[self performAsCurrentScheduler:block];
	});

	return disposable;
}

-schedule 在 self.queue 队列中异步调用了 -performAsCurrentScheduler

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
- (RACDisposable *)after:(NSDate *)date schedule:(void (^)(void))block {
	NSCParameterAssert(date != nil);
	NSCParameterAssert(block != NULL);

	RACDisposable *disposable = [[RACDisposable alloc] init];

	dispatch_after([self.class wallTimeWithDate:date], self.queue, ^{
		if (disposable.disposed) return;
		[self performAsCurrentScheduler:block];
	});

	return disposable;
}

+ (dispatch_time_t)wallTimeWithDate:(NSDate *)date {
	NSCParameterAssert(date != nil);

	double seconds = 0;
	double frac = modf(date.timeIntervalSince1970, &seconds);

	struct timespec walltime = {
		.tv_sec = (time_t)fmin(fmax(seconds, LONG_MIN), LONG_MAX),
		.tv_nsec = (long)fmin(fmax(frac * NSEC_PER_SEC, LONG_MIN), LONG_MAX)
	};

	return dispatch_walltime(&walltime, 0);
}
  1. 调用 wallTimeWithDate 方法将 NSDate 转化成 dispatch_time_t
  2. 调用 dispatch_after 将 block 放进队列 self.queue 中延迟执行
  3. block 执行过程中是通过 -performAsCurrentScheduler: 方法触发，触发前判断 disposable 是否被取消，取消则直接返回
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
- (RACDisposable *)after:(NSDate *)date repeatingEvery:(NSTimeInterval)interval withLeeway:(NSTimeInterval)leeway schedule:(void (^)(void))block {
	NSCParameterAssert(date != nil);
	NSCParameterAssert(interval > 0.0 && interval < INT64_MAX / NSEC_PER_SEC);
	NSCParameterAssert(leeway >= 0.0 && leeway < INT64_MAX / NSEC_PER_SEC);
	NSCParameterAssert(block != NULL);

	uint64_t intervalInNanoSecs = (uint64_t)(interval * NSEC_PER_SEC);
	uint64_t leewayInNanoSecs = (uint64_t)(leeway * NSEC_PER_SEC);

	dispatch_source_t timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, self.queue);
	dispatch_source_set_timer(timer, [self.class wallTimeWithDate:date], intervalInNanoSecs, leewayInNanoSecs);
	dispatch_source_set_event_handler(timer, block);
	dispatch_resume(timer);

	return [RACDisposable disposableWithBlock:^{
		dispatch_source_cancel(timer);
	}];
}

该方法是通过 GCD 创建定时任务，然后通过 dispatch_source_set_event_handler 把参数 block 和 计时器关联起来，任务被取消的时候取消对应的定时器

RACTargetQueueScheduler

RACTargetQueueScheduler 继承于 RACQueueScheduler，提供了一新的初始化方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
- (instancetype)initWithName:(NSString *)name targetQueue:(dispatch_queue_t)targetQueue {
	NSCParameterAssert(targetQueue != NULL);

	if (name == nil) {
		name = [NSString stringWithFormat:@"org.reactivecocoa.ReactiveObjC.RACTargetQueueScheduler(%s)", dispatch_queue_get_label(targetQueue)];
	}

	dispatch_queue_t queue = dispatch_queue_create(name.UTF8String, DISPATCH_QUEUE_SERIAL);
	if (queue == NULL) return nil;

	dispatch_set_target_queue(queue, targetQueue);

	return [super initWithName:name queue:queue];
}

在初始化方法创建了串行队列 queue，然后通过 dispatch_set_target_queue 把 targetQueue 和 queue 关联起来

dispatch_set_target_queue 有2个作用:

  1. 初始化方法内 queue 是通过 dispatch_queue_create 创建，无法设置优先级，dispatch_set_target_queue 可以将 queue 优先级设置为 targetQueue 的优先级

  2. 设置队列的层次体系，可以理解为 queue 中的任务会派发给 targetQueue；比如如果 targetQueue 是 DISPATCH_QUEUE_SERIAL 串行队列，则 queue 中的任务也是串行执行；如果多个 queue 都指定同一个 targetQueue 串行队列，那么多个 queue 的任务是同步执行的

    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
    - (void)testSetTarget {
        dispatch_queue_t targetQueue = dispatch_queue_create("targetQueue", DISPATCH_QUEUE_SERIAL);
        dispatch_queue_t queue1 = dispatch_queue_create("queue1", DISPATCH_QUEUE_SERIAL);
        dispatch_queue_t queue2 = dispatch_queue_create("queue2", DISPATCH_QUEUE_CONCURRENT);
        
        dispatch_set_target_queue(queue1, targetQueue);
        dispatch_set_target_queue(queue2, targetQueue);
        
        dispatch_async(queue1, ^{
            NSLog(@"1. queue1 excute");
        });
        dispatch_async(queue1, ^{
            [NSThread sleepForTimeInterval:1.f];
            NSLog(@"2. queue1 excute");
        });
        dispatch_async(queue2, ^{
            NSLog(@"1. queue2 excute");
        });
        dispatch_async(queue2, ^{
            NSLog(@"2. queue2 excute");
        });
        dispatch_async(targetQueue, ^{
            NSLog(@"target queue");
        });
    }

    输出：

    1
    2
    3
    4
    5
    2019-03-31 19:47:32.441322+0800 AppTest[42843:1630586] 1. queue1 excute
    2019-03-31 19:47:33.446694+0800 AppTest[42843:1630586] 2. queue1 excute
    2019-03-31 19:47:33.446917+0800 AppTest[42843:1630586] 1. queue2 excute
    2019-03-31 19:47:33.447003+0800 AppTest[42843:1630586] 2. queue2 excute
    2019-03-31 19:47:33.447095+0800 AppTest[42843:1630586] target queue

RACSubscriptionScheduler

相对于父类，主要添加了一个私有的属性:

1
@property (nonatomic, strong, readonly) RACScheduler *backgroundScheduler;

在其初始化的时候会创建 backgroundScheduler

1
2
3
4
5
6
7
- (instancetype)init {
	self = [super initWithName:@"org.reactivecocoa.ReactiveObjC.RACScheduler.subscriptionScheduler"];

	_backgroundScheduler = [RACScheduler scheduler];

	return self;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
- (RACDisposable *)schedule:(void (^)(void))block {
	NSCParameterAssert(block != NULL);

	if (RACScheduler.currentScheduler == nil) return [self.backgroundScheduler schedule:block];

	block();
	return nil;
}

- (RACDisposable *)after:(NSDate *)date schedule:(void (^)(void))block {
	RACScheduler *scheduler = RACScheduler.currentScheduler ?: self.backgroundScheduler;
	return [scheduler after:date schedule:block];
}

- (RACDisposable *)after:(NSDate *)date repeatingEvery:(NSTimeInterval)interval withLeeway:(NSTimeInterval)leeway schedule:(void (^)(void))block {
	RACScheduler *scheduler = RACScheduler.currentScheduler ?: self.backgroundScheduler;
	return [scheduler after:date repeatingEvery:interval withLeeway:leeway schedule:block];
}

其他的方法大体逻辑都是判断当前线程有没有对应的 RACScheduler，如果有任务则在当前线程对应的 RACScheduler 执行，若没有则在 backgroundScheduler 上执行