ReactiveCocoa------冷信号vs热信号

发表于 | 分类于

冷热信号的概念是源自于源于.NET框架Reactive Extensions(RX)中的Hot Observable和Cold Observable，两者的区别是：

  1. Hot Observable是主动的，尽管你并没有订阅事件，但是它会时刻推送，就像鼠标移动；而Cold Observable是被动的，只有当你订阅的时候，它才会发布消息。
  2. Hot Observable可以有多个订阅者，是一对多，集合可以与订阅者共享信息；而Cold Observable只能一对一，当有不同的订阅者，消息是重新完整发送。

冷热信号分别对应RAC中的RACSignal和RACSubject，RACSubject是RACSignal的子类。本文中笔者将从示例代码引入RACSignal（冷信号）在实际应用中的问题及RACSubject（热信号）的使用，通过剖析RACSubject的源码，阐述冷热信号的概念及RACSubject的实现。

懒惰的RACSignal

笔者函数式编程提到了惰性求值的概念:

惰性求值（尽可能延迟表达式求值），表达式不会在它被绑定到变量之后就立即求值，而是在该值被取用的时候求值。

RACSignal就是惰性求值的。RACSignal的didSubscribe block只有在RACSignal被订阅的时候才会被执行，而且每次订阅RACSignal都会执行一遍didSubscribe block。也正是这个特性赋予了RACSignal冷信号的特点。同时，RACSignal的这些特点也导致了其在实际应用中的一些问题。

举个栗子

一起看以下示例代码

示例一：RACSignal & 副作用

以下代码在didSubscribe中引入副作用”i += 1“，通过Output可以看出这种情况下对同一个信号多次订阅拿到不同的值，这显然是我们不想看到的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
__block int i = 0;
RACSignal * signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
    i += 1;
    [subscriber sendNext:@(i)];
    [subscriber sendCompleted];
    return nil;
}];

[signal subscribeNext:^(id x) {
    NSLog(@"subscriber1---recived---%@",x);
}];
[signal subscribeNext:^(id x) {
    NSLog(@"subscriber2---recived---%@",x);
}];
[signal subscribeNext:^(id x) {
    NSLog(@"subscriber3---recived---%@",x);
}];

Output

1
2
3
2018-03-04 18:53:38.660673+0800 RACDemo[98599:6473410] subscriber1---recived---1
2018-03-04 18:53:38.661029+0800 RACDemo[98599:6473410] subscriber2---recived---2
2018-03-04 18:53:38.661184+0800 RACDemo[98599:6473410] subscriber3---recived---3

RACSignal & 时间

以下代码signal以1秒的间隔依次发送1，2，3。subscriber1立即订阅signal（signal未发送任何值之前），subscribe2在3.1秒后（signal将所有值发送完毕后）订阅signal。在不同时间订同一信号获取的结果是一样的。通过Output可以看出在任意时间点订阅signal，signal都会以1秒的间隔依次发送1，2，3。很多情况下，这同样不是我们想看到的，subscribe2只关心signal在3.1秒之后发送的值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
RACSignal * signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
    [[RACScheduler mainThreadScheduler] afterDelay:1 schedule:^{
        [subscriber sendNext:@1];
    }];
    [[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
        [subscriber sendNext:@2];
    }];
    [[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
        [subscriber sendNext:@3];
    }];
    return nil;
}];

[signal subscribeNext:^(id x) {
    NSLog(@"subscribe1----recieve-%@",x);
}];


[[RACScheduler mainThreadScheduler] afterDelay:3.1 schedule:^{
    [signal subscribeNext:^(id x) {
        NSLog(@"subscribe2----recieve-%@",x);
    }];
}];

Output

注意时间 

1
2
3
4
5
6
2018-03-04 18:58:21.864222+0800 RACDemo[98639:6481792] subscribe1----recieve-1
2018-03-04 18:58:22.863358+0800 RACDemo[98639:6481792] subscribe1----recieve-2
2018-03-04 18:58:23.963592+0800 RACDemo[98639:6481792] subscribe1----recieve-3
2018-03-04 18:58:24.967656+0800 RACDemo[98639:6481792] subscribe2----recieve-1
2018-03-04 18:58:25.964144+0800 RACDemo[98639:6481792] subscribe2----recieve-2
2018-03-04 18:58:26.967526+0800 RACDemo[98639:6481792] subscribe2----recieve-3

结合在懒惰的RACSignal中提到RACSignal的特点。不难理解，示例一、示例二所展示的结果正式由于RACSignal的惰性求值及每次订阅重复调用didSubscribe block导致的。

RACSubject

我们对示例二的代码进行一个小的改动,创建热信号subject,subject订阅signal，subscriber订阅subject不再直接订阅signal。

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
RACSignal * signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
    [[RACScheduler mainThreadScheduler] afterDelay:1 schedule:^{
        [subscriber sendNext:@1];
    }];
    [[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
        [subscriber sendNext:@2];
    }];
    [[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
        [subscriber sendNext:@3];
    }];
    return nil;
}];

RACSubject * subject = [RACSubject subject];
[signal subscribe:subject];


[subject subscribeNext:^(id x) {
    NSLog(@"subject subscriber1----recieve-%@",x);
}];


[[RACScheduler mainThreadScheduler] afterDelay:3.1 schedule:^{
    [subject subscribeNext:^(id x) {
        NSLog(@"subject subscriber2----recieve-%@",x);
    }];
}];

Output

一起来看看发生了啥，立即订阅的subscriber1收到了所有值，而3.1秒之后订阅的subscribe2没有收到任何值。RACSubject就是RAC中的热信号，subject字面意思是“主题”。RACSubject是与时间强相关的。在事件发生时，只有已经订阅该主题的subscriber才会被通知。

1
2
3
2018-03-05 13:30:38.591185+0800 RACDemo[11918:7623749] subject subscriber1----recieve-1
2018-03-05 13:30:39.590616+0800 RACDemo[11918:7623749] subject subscriber1----recieve-2
2018-03-05 13:30:40.689946+0800 RACDemo[11918:7623749] subject subscriber1----recieve-3

RACSubject的实现

我们通过分析RACSubject的源码来探究热信号的概念。

RACSubject是RACSignal的子类。相比于RACSignal丰富的头文件，RACSubject对外的接口并没有提供太多方法：

1
2
3
@interface RACSubject : RACSignal <RACSubscriber>
+ (instancetype)subject;
@end

RACSubject的特性主要由subscribers订阅者数组和RACSubscriber协议提供。

subscribers订阅者数组

RACSubject维护了一个订阅者数组，每当有新的订阅者产生，都会将传入的 id 对象加入数组：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
        RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
        subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

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

订阅的过程分为三个部分：

  1. 初始化一个 RACPassthroughSubscriber 实例；
  2. 将subscriber加入RACSubject持有的数组中；
  3. 创建一个RACDisposable对象，在当前subscriber销毁时，将自身从数组中移除。

订阅者数组，为RACSubject提供了一对多的能力。事件发生时RACSubject通过实现RACSubscriber协议提供的方法，遍历subscribers数组逐个发送消息。

RACSubscriber协议

RACSubscriber协议为RACSubject提供了在创建成功后向订阅者继续发送消息的能力，RACSignal只能通过创建信号的didSubscribe block遵循该协议的subscriber发送消息。所以我们可以说，RACSignal是不可变的，RACSubject是可变的。

1
2
3
4
5
6
7
8
9
@protocol RACSubscriber <NSObject>
@required

- (void)sendNext:(nullable id)value;
- (void)sendError:(nullable NSError *)error;
- (void)sendCompleted;
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable;

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

- (void)sendError:(NSError *)error {
        [self.disposable dispose];

        [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
                [subscriber sendError:error];
        }];
}

- (void)sendCompleted {
        [self.disposable dispose];

        [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
                [subscriber sendCompleted];
        }];
}

RACSubject实现相对简单。类似于通知，通过维护订阅者数组在事件发生后为所有订阅者发送消息。在具体的编码过程中加入了一些锁用以避免线程竞争。

总结

至此，结合Hot Observable和Cold Observable。我们不难理解，冷信号就好像看录像，热信号就好像是看直播。

冷信号（看录像）
不可变：RACSignal创建后，将要发送的消息就是固定的。（录像录制完成后内容就是固定的）
一对一：当有不同的订阅者，RACSignal会重复发送所有消息。(为每一个观看者重复播放录像)
被动的：只有订阅者订阅的时候，才会发送消息。（没有人要求观看时不会播放录像）

热信号（看直播）
可变：RACSubject创建后，可以持续添加新事件。（直播总会有新的事情发生）
一对多：所有订阅者共享同一个RACSubject。（只有一个主播）
主动的：无论是否有订阅者，都会发送新事件。（有没有人看主播都会播）

相关链接

细说ReactiveCocoa的冷信号与热信号
Comparing replay, replayLast, and replayLazily