RAC 学习探究 二、创建订阅信号流程

[TOC]

-w814

iOS框架_RAC01_signalCreate@40

Subscriber like todo ： 订阅者的操作

订阅者 Subscriber 可以调用三个常见的方法，sendNext，sendError，sendComplete

sendNext可以发送很多次，但是直到遇到了Error或者Complete之后就结束了。

这里可以查看三个方法的源码，得到这个的证明：

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
#pragma mark RACSubscriber

- (void)sendNext:(id)value {
	@synchronized (self) {
		void (^nextBlock)(id) = [self.next copy];
		if (nextBlock == nil) return;
		nextBlock(value);
	}
}

- (void)sendError:(NSError *)e {
	@synchronized (self) {
		void (^errorBlock)(NSError *) = [self.error copy];
		[self.disposable dispose]; // 先销毁，再调用 errorBlock

		if (errorBlock == nil) return;
		errorBlock(e);
	}
}

- (void)sendCompleted {
	@synchronized (self) {
		void (^completedBlock)(void) = [self.completed copy];
		[self.disposable dispose]; // 先销毁，再调用 completedBlock

		if (completedBlock == nil) return;
		completedBlock();
	}
}

RACPassthroughSubscriber 核心穿透订阅者

订阅的流程，调用的 RACSignal 类中的 subscriptNext 方法，其实是调用类本身的另外一个方法：subscribe：subscriber方法，实现如下：

详见上图中有这个方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
	NSCParameterAssert(subscriber != nil);

//    Dispose 销毁者 最后返回销毁者
	RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
	subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

	if (self.didSubscribe != NULL) {
		RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
//            调用了 didSubscriber 
			RACDisposable *innerDisposable = self.didSubscribe(subscriber);
			[disposable addDisposable:innerDisposable];
		}];

		[disposable addDisposable:schedulingDisposable];
	}
	
	return disposable;
}

这里有一个核心类：RACPassthroughSubscriber，我把他称之为穿透订阅者。

穿透订阅者的这个生成方法，我们一看传入了三个参数：subscriber，signal，disposable，有了这桑个参数，是不是基本上可以重写RAC了，所以这个类很关键，可以说它是核心穿透订阅者了。

RACPassthroughSubscriber 的 init 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
- (instancetype)initWithSubscriber:(id<RACSubscriber>)subscriber
                            signal:(RACSignal *)signal
                        disposable:(RACCompoundDisposable *)disposable {
    // 核心穿透订阅者 的init 入参 非常全面，
//    有： subscriber signal disposable
    // 这三个参数连起来念，订阅信号销毁，它大概就是做这个的。
	NSCParameterAssert(subscriber != nil);

	self = [super init];

	_innerSubscriber = subscriber;
	_signal = signal;
	_disposable = disposable;

	[self.innerSubscriber didSubscribeWithDisposable:self.disposable];
	return self;
}

最后给 _innerSubscriber 其实也就是 订阅者 subscriber 关联了一个回调方法，在销毁的时候互相回调使用的：

1
2
3
4
5
6
7
8
9
10
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)d {
	if (d.disposed) return;
	[self.disposable addDisposable:d];

	@weakify(self, d);
	[d addDisposable:[RACDisposable disposableWithBlock:^{
		@strongify(self, d);
		[self.disposable removeDisposable:d];
	}]];
}

调度者 RACSchedule

Schedule 里面全部都是多线程相关的操作

SRCSubscriptionScheduler 是一个单例对象，订阅调度者。

    RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:

这个方法调用的时候，调用了 RACSubscriptionScheduler.m 的 schedule 方法：

RACSubscriptionScheduler.m ———— schedule

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

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

	block();
	return nil;
}

它会调用 scheduler 的 currentScheduler 判断：

1
2
3
4
5
6
7
8
9
10
+ (RACScheduler *)currentScheduler {
    // 从当前线程的线程字典里面 tls 里面取一个字符串的value，
//    第一次肯定为nil没有值，然后就设置成一个
	RACScheduler *scheduler = NSThread.currentThread.threadDictionary[RACSchedulerCurrentSchedulerKey];
	if (scheduler != nil) return scheduler;
    // 如果在主线程上，就默认返回主线程，否则返回nil
	if ([self.class isOnMainThread]) return RACScheduler.mainThreadScheduler;

	return nil;
}

所以这个代码有三种情况：

  1. 主线程第一次进入：当前线程的TLS里面，没有这个key对应的值，判断是否在主线程，在的话返回主线程的 mainThreadScheduler
  2. 子线程第一次进入：…不在主线程，返回nil
  3. 第二次进入，这个表里面有值了，之前子线程进来的时候下面章节填充上的，返回取出来的scheduler。存值的实际后面讲。

发生了第二个情况的话，返回的是nil，RACSubscriptionScheduler.m ———— schedule里面就会继续调用一个属性叫做 backgroundScheduler

这个backgroundScheduler 的声明路程是这样的：

  1. init初始化的时候他被init [RACScheduler scheduler]
  2. scheduler 这个方法返回了：[self schedulerWithPriority:RACSchedulerPriorityDefault]
  3. [self schedulerWithPriority:priority name:@"org.reactivecocoa.ReactiveObjC.RACScheduler.backgroundScheduler"];
  4. 板凳君自己创建了一个全局并发队列:[[RACTargetQueueScheduler alloc] initWithName:name targetQueue:dispatch_get_global_queue(priority, 0)];
    解释一下参数，优先级是上面第二点传递过来的RACSchedulerPriorityDefault
  5. 这个方法的实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
- (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];
}

这里可以看到，串行一个队列，串行子线程操作它保证异步，但是调度的时候串行来入队
牺牲了信号的并发，保证调度的有序。

所以这个backgroundScheduler 就是一个自定义的串行队列

那么这个自定义串行上调用了 schedule 方法，实现如下：

RACQueueScheduler.m ———— schedule

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;
}

performAsCurrentScheduler

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
- (void)performAsCurrentScheduler:(void (^)(void))block {
	NSCParameterAssert(block != NULL);

//    如果我们正在使用并发队列，
//    那么我们可以并发地结束在这里， 在这种情况下，
//    我们*不想*希望在块完成执行后立即清除当前调度程序，
//    而只能*在*所有并发之后 调用完成。
	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];
	}
}

把 self 自己，放进刚刚的那个线程里面的TLS的字典里面去，下次很方便。执行Block。更新字典表（目前是板凳调度者，如果有previous也就是主体的那个调度者，就把主体调度者放进字典表）。

这里操作了字典表，当前线程的TLS里面也有值了，所以第三点也成立了，子线程进来之后直接取值出调度者，直接执行了block，不用再调用 backgroundScheduler 的方法了

RACObserve 原理

试用的第一个Demo，给copy过来，凑一下字数和排版

1
2
3
4
// count 是 self 的 @property
    [RACObserve(self, count) subscribeNext:^(id  _Nullable x) {
        NSLog(@"new count value : %@",x);
    }];

RACObserve 的原理，这个是一个宏，两句话可以说明白：

第一句话：观察体封装成了一个对象 RACKVOTrampoline，这一通操作，返回 RACSignal。

第二句话：接收观察的中间代理 proxy 是个全局单例，接收所有的 observer 回调，进行派发。

remove 的原理：RACKVOTrampoline的析构，delloc 里面 remove 所有的观察。

RACSubject 全能订阅者

单独使用它既可以发送也可以订阅，但是记住先订阅再发送，demo如下：

1
2
3
4
5
RACSubject *sub = [RACSubject subject];
[sub subscribeNext:^(id  _Nullable x) {
            NSLog(@"%@", x);
}];
[sub sendNext:@"打印出来的字符串"];

整个流程跟上面篇章里面的订阅流程类似。

Subject 实现自主收发的原理

三点：

  1. RACSubject 在工厂方法初始化的时候，内部创建了一个订阅者的数组。
  2. 在 subscribeNext 的时候，进行了把刚创建的核心穿透订阅者添加到数组的操作。
  3. 在sendNext的时候，遍历了这个数组，挨个发送了block。

根据这三点扒一下源代码：

1. RACSubjext init 时初始化数组：

1
2
3
4
5
6
7
8
9
- (instancetype)init {
	self = [super init];
	if (self == nil) return nil;

	_disposable = [RACCompoundDisposable compoundDisposable];
	_subscribers = [[NSMutableArray alloc] initWithCapacity:1];
	
	return self;
}

2. 核心穿透订阅者 创建后就被添加进数组

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
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
	NSCParameterAssert(subscriber != nil);

	RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
	subscriber = [[RACPassthroughSubscriber alloc]
                  initWithSubscriber:subscriber
                    signal:self
                    disposable:disposable];

    //把 核心穿透订阅者 添加到了 订阅者数组里面了
    //后面发送信号的实质就是，遍历这个subscribers数组发送就可以了
	NSMutableArray *subscribers = self.subscribers;
	@synchronized (subscribers) {
		[subscribers addObject:subscriber];
	}
	
	[disposable addDisposable:[RACDisposable disposableWithBlock:^{
		@synchronized (subscribers) {
			// 由于较新的订户的寿命通常较短
            // 所以搜索从列表的末尾开始
			NSUInteger index = [subscribers
                                indexOfObjectWithOptions:NSEnumerationReverse
                                passingTest:^ BOOL (id<RACSubscriber> obj,
                                                    NSUInteger index,
                                                    BOOL *stop) {
				return obj == subscriber;
			}];
			if (index != NSNotFound) [subscribers removeObjectAtIndex:index];
		}
	}]];
	return disposable;
}

3. sendNext遍历数组，发送 Block

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
- (void)sendNext:(id)value {
	[self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
		[subscriber sendNext:value];
	}];
}

- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {
	NSArray *subscribers;
	@synchronized (self.subscribers) {
		subscribers = [self.subscribers copy];
	}
    
//    就是一个for循环，遍历调用
	for (id<RACSubscriber> subscriber in subscribers) {
		block(subscriber);
	}
}

RACMulticastConnection

工厂的方法是publish，调用connect.subscribeNext 不会重复生成。

RAC 宏定义

RAC 的宏在预编译阶段，是一个特别宏大的系统，他提个的是一个宏定义系统，可以单独研究很久很久。。。。。。

@weakify

@ 的作用是从C语言编译到OC，例如字符串。
weakify最终的效果就是

__weak __typeof__(self) self_weak_ = self;
赏

谢谢 请给我一支兰州

支付宝

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

把我的全部，<br>奉献给我所热爱的一切。<br><br>谢谢大家 ~