硬核干货——《中小企业 AI 实战指南》免费下载! 了解详情
写点什么

可变的热信号 RACSubject

  • 2019-12-09
  • 本文字数:6400 字

    阅读完需:约 21 分钟

可变的热信号 RACSubject

在 ReactiveCocoa 中除了不可变的信号 RACSignal,也有用于桥接非 RAC 代码到 ReactiveCocoa 世界的『可变』信号 RACSubject



RACSubject 到底是什么?根据其字面意思,可以将它理解为一个可以订阅的主题,我们在订阅主题之后,向主题发送新的消息时,所有的订阅者都会接收到最新的消息。


但是这么解释确实有点晦涩,也不易于理解,ReactiveCocoa 团队对 RACSubject 的解释是,RACSubject 其实就是一个可以手动控制的信号(感觉这么解释更难理解了)。


A subject, represented by the RACSubject class, is a signal that can be manually controlled.

RACSubject 简介

RACSubjectRACSignal 的子类,与 RACSignal 以及 RACSequence 有着众多的类簇不同,RACSubject 在整个工程中并没有多少子类;不过,在大多数情况下,我们也只会使用 RACSubject 自己或者 RACReplaySubject



相比于 RACSignal 丰富的头文件 ,RACSubject 对外的接口并没有提供太多的方法:


Objective-C


@interface RACSubject : RACSignal <RACSubscriber>
+ (instancetype)subject;
@end
复制代码


唯一提供的接口就是用于返回一个新实例的 +subject 方法;除此之外,在笔者看来它与 RACSignal 最大的不同就是:RACSubject 实现了 RACSubscriber 协议,也就是下面的这些方法:


Objective-C


@protocol RACSubscriber <NSObject>@required
- (void)sendNext:(nullable id)value;- (void)sendError:(nullable NSError *)error;- (void)sendCompleted;- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable;
@end
复制代码


我们并不能在一个 RACSignal 对象上执行这些方法,只能在创建信号的 block 里面向遵循 RACSubscriber 协议的对象发送新的值或者错误,这也是 RACSubject 和父类最大的不同:在 RACSubject 实例初始化之后,也可以通过这个实例向所有的订阅者发送消息。

冷信号与热信号

提到 RACSubject 就不得不提 ReactiveCocoa 中的另一对概念,冷信号和热信号。


其实解释这两者之间区别的文章已经很多了,我相信各位读者能找到很多的资料,在这里就简单介绍一下冷热信号的概念,如果想要了解更多的内容可以在 References 中找到更多的文章。


对于冷热信号概念,我们借用 Rx 中的描述:



Cold signal is sequences that are passive and start producing notifications on request (when subscribed to), and hot signal is sequences that are active and produce notifications regardless of subscriptions. —- Hot and Cold observables


冷信号是被动的,只会在被订阅时向订阅者发送通知;热信号是主动的,它会在任意时间发出通知,与订阅者的订阅时间无关。


也就是说冷信号所有的订阅者会在订阅时收到完全相同的序列;而订阅热信号之后,只会收到在订阅之后发出的序列。


热信号的订阅者能否收到消息取决于订阅的时间。


热信号在我们生活中有很多的例子,比如订阅杂志时并不会把之前所有的期刊都送到我们手中,只会接收到订阅之后的期刊;而对于冷信号的话,举一个不恰当的例子,每一年的高考考生在『订阅』高考之后,收到往年所有的试卷,并在高考之后会取消订阅。

热信号 RACSubject

在 ReactiveCocoa 中,我们使用 RACSignal 来表示冷信号,也就是每一个订阅者在订阅信号时都会收到完整的序列;RACSubject 用于表示热信号,订阅者接收到多少值取决于它订阅的时间。


前面的文章中已经对 RACSignal 冷信号有了很多的介绍,这里也就不会多说了;这一小节主要的内容是想通过一个例子,简单展示 RACSubject 的订阅者收到的内容与订阅时间的关系:


Objective-C


RACSubject *subject = [RACSubject subject];
// Subscriber 1[subject subscribeNext:^(id _Nullable x) { NSLog(@"1st Sub: %@", x);}];[subject sendNext:@1];
// Subscriber 2[subject subscribeNext:^(id _Nullable x) { NSLog(@"2nd Sub: %@", x);}];[subject sendNext:@2];
// Subscriber 3[subject subscribeNext:^(id _Nullable x) { NSLog(@"3rd Sub: %@", x);}];[subject sendNext:@3];[subject sendCompleted];
复制代码


这里以图的方式来展示整个订阅与订阅者接收消息的过程:



从图中我们可以清楚的看到,几个订阅者根据订阅时间的不同收到了不同的数字序列,RACSubject时间相关的,它在发送消息时只会向已订阅的订阅者推送消息。

RACSubject 的实现

RACSubject 的实现并不复杂,它『可变』的特性都来源于持有的订阅者数组 subscribers,在每次执行 subscribeNext:error:completed: 一类便利方法时,都会将传入的 id<RACSubscriber> 对象加入数组:


Objective-C


- (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 销毁时,将自身从数组中移除。



-subscribe: 将所有遵循 RACSubscriber 协议的对象全部加入当前 RACSubject 持有的数组 subscribers 中。


在上一节的例子中,我们能对 RACSubject 发送 -sendNext: 等消息也都取决于它实现了 RACSubscriber 协议:


Objective-C


- (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 会在自身接受到这些方法时,下发给持有的全部的 subscribers



代码中的 -enumerateSubscribersUsingBlock: 只是一个使用 for 循环遍历 subscribers 的安全方法:


Objective-C


- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {  NSArray *subscribers;  @synchronized (self.subscribers) {    subscribers = [self.subscribers copy];  }
for (id<RACSubscriber> subscriber in subscribers) { block(subscriber); }}
复制代码


RACSubject 就是围绕一个 NSMutableArray 数组实现的,实现还是非常简单的,只是在需要访问 subscribers 的方法中使用 @synchronized 避免线程竞争。


Objective-C


@interface RACSubject ()
@property (nonatomic, strong, readonly) NSMutableArray *subscribers;
@end
复制代码


RACSubject 提供的初始化类方法 +subject 也只是初始化了几个成员变量:


Objective-C


+ (instancetype)subject {  return [[self alloc] init];}
- (instancetype)init { self = [super init]; if (self == nil) return nil;
_disposable = [RACCompoundDisposable compoundDisposable]; _subscribers = [[NSMutableArray alloc] initWithCapacity:1];
return self;}
复制代码


至此,对于 RACSubject 的分析就结束了,接下来会分析更多的子类。

RACBehaviorSubject 与 RACReplaySubject

这一节会介绍 RACSubject 的两个子类 RACBehaviorSubjectRACReplaySubject,前者在订阅时会向订阅者发送最新的消息,后者在订阅之后可以重新发送之前的所有消息序列。

RACBehaviorSubject

先来介绍两者中实现较简单的 RACBehaviorSubject,它在内部会保存一个 currentValue 对象,也就是最后一次发送的消息:


Objective-C


@interface RACBehaviorSubject ()
@property (nonatomic, strong) id currentValue;
@end
复制代码


在每次执行 -sendNext: 时,都会对 RACBehaviorSubject 中保存的 currentValue 进行更新,并使用父类的 -sendNext: 方法,向所有的订阅者发送最新的消息:


Objective-C


- (void)sendNext:(id)value {  @synchronized (self) {    self.currentValue = value;    [super sendNext:value];  }}
复制代码


RACBehaviorSubject 最重要的特性就是在订阅时,向最新的订阅者发送之前的消息,这是通过覆写 -subscribe: 方法实现的。


在调用子类的 -subscribe: 方法之后,会在 subscriber 对象上执行 -sendNext: 方法:


Objective-C


- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {  RACDisposable *subscriptionDisposable = [super subscribe:subscriber];
RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{ @synchronized (self) { [subscriber sendNext:self.currentValue]; } }];
return [RACDisposable disposableWithBlock:^{ [subscriptionDisposable dispose]; [schedulingDisposable dispose]; }];}
复制代码


接下来,通过一个简单的例子来演示 RACBehaviorSubject 到底是如何工作的:


Objective-C


RACBehaviorSubject *subject = [RACBehaviorSubject subject];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"1st Sub: %@", x);}];[subject sendNext:@1];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"2nd Sub: %@", x);}];[subject sendNext:@2];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"3rd Sub: %@", x);}];[subject sendNext:@3];[subject sendCompleted];
复制代码


上面的代码其实与 RACSubject 一节中的代码差不多,只将 RACSubject 转换成了 RACBehaviorSubject 对象。



在每次订阅者订阅 RACBehaviorSubject 之后,都会向该订阅者发送最新的消息,这也就是 RACBehaviorSubject 最重要的行为。


RACBehaviorSubject 有一个用于创建包含默认值的类方法 +behaviorSubjectWithDefaultValue:,如果将上面的第一行代码改成:


Objective-C


RACBehaviorSubject *subject = [RACBehaviorSubject behaviorSubjectWithDefaultValue:@0];
复制代码


那么在第一个订阅者刚订阅 RACBehaviorSubject 时就会收到 @0 对象。


RACReplaySubject

RACReplaySubject 相当于一个自带 bufferRACBehaviorSubject,它可以在每次有新的订阅者订阅之后发送之前的全部消息。


Objective-C


@interface RACReplaySubject ()
@property (nonatomic, assign, readonly) NSUInteger capacity;@property (nonatomic, strong, readonly) NSMutableArray *valuesReceived;
@end
复制代码


实现的方式是通过持有一个 valuesReceived 的数组和能够存储的对象的上限 capacity,默认值为:


Objective-C


const NSUInteger RACReplaySubjectUnlimitedCapacity = NSUIntegerMax;
复制代码


当然你可以用 +replaySubjectWithCapacity: 初始化一个其它大小的 RACReplaySubject 对象:


Objective-C


+ (instancetype)replaySubjectWithCapacity:(NSUInteger)capacity {  return [(RACReplaySubject *)[self alloc] initWithCapacity:capacity];}
- (instancetype)initWithCapacity:(NSUInteger)capacity { self = [super init];
_capacity = capacity; _valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
return self;}
复制代码


在每次调用 -sendNext: 方法发送消息时,都会将其加入 valuesReceived 数组中,并踢出之前的元素:


Objective-C


- (void)sendNext:(id)value {  @synchronized (self) {    [self.valuesReceived addObject:value ?: RACTupleNil.tupleNil];    [super sendNext:value];
if (self.capacity != RACReplaySubjectUnlimitedCapacity && self.valuesReceived.count > self.capacity) { [self.valuesReceived removeObjectsInRange:NSMakeRange(0, self.valuesReceived.count - self.capacity)]; } }}
复制代码


需要注意的有两点,一是对 valuesReceived 的数组的操作必须使用 @synchronized 加锁;第二,如果 value 为空的话,也需要将其转换成 RACTupleNil.tupleNil 对象进行保存。



-sendError:-sendCompleted 方法都会标记对应 flag,即 hasCompletedhasError,这里就不介绍了;同样的,RACReplaySubject 也覆写了 -subscribe: 方法,在每次有订阅者订阅时重新发送所有的序列:


Objective-C


- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {  RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{ @synchronized (self) { for (id value in self.valuesReceived) { if (compoundDisposable.disposed) return;
[subscriber sendNext:(value == RACTupleNil.tupleNil ? nil : value)]; }
if (compoundDisposable.disposed) return;
if (self.hasCompleted) { [subscriber sendCompleted]; } else if (self.hasError) { [subscriber sendError:self.error]; } else { RACDisposable *subscriptionDisposable = [super subscribe:subscriber]; [compoundDisposable addDisposable:subscriptionDisposable]; } } }];
[compoundDisposable addDisposable:schedulingDisposable];
return compoundDisposable;}
复制代码


我们仍然使用上一节中的例子来展示 RACReplaySubject 是如何工作的,只修改第一行代码:


Objective-C


RACReplaySubject *subject = [RACReplaySubject subject];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"1st Subscriber: %@", x);}];[subject sendNext:@1];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"2nd Subscriber: %@", x);}];[subject sendNext:@2];
[subject subscribeNext:^(id _Nullable x) { NSLog(@"3rd Subscriber: %@", x);}];[subject sendNext:@3];[subject sendCompleted];
复制代码


运行这段代码之后,会得到如下图的结果:



所有订阅 RACReplaySubject 的对象(默认行为)都能获得完整的序列,而这个特性在与 RACMulticastConnection 一起使用也有着巨大威力,我们会在之后的文章中介绍。

总结

RACSubjectRACSignal 对象之上进行了简单的修改,将原有的冷信号改造成了热信号,将不可变变成了可变。


虽然 RACSubject 的实现并不复杂,只是存储了一个遵循 RACSubscriber 协议的对象列表以及所有的消息,但是在解决实际问题时却能够很好地解决很多与网络操作相关的问题。

References


Github Repo:iOS-Source-Code-Analyze


Source: https://draveness.me/racsubject

关于图片和转载

本作品采用知识共享署名 4.0 国际许可协议进行许可。


  转载时请注明原文链接,图片在使用时请保留图片中的全部内容,可适当缩放并在引用处附上图片所在的文章链接,图片使用 Sketch 进行绘制,你可以在 [](https://draveness.me/draveness.me/sketch-sketch) 一文中找到画图的方法和素材。
复制代码


本文转载自 Draveness 技术博客。


原文链接:https://draveness.me/racsubject


2019-12-09 15:56774

评论

发布
暂无评论
发现更多内容

887页Java面试“成神”手册,已助朋友狂砍9个一二线大厂Offer

Java架构追梦

Java 阿里巴巴 架构 面试 金三银四

智慧党建管理系统,智慧组工平台开发方案

13530558032

一线互联网大厂面经分享:阿里三面+头条四面+腾讯二面+美团四面

Java架构之路

Java 程序员 架构 面试 编程语言

接口测试--apipost中cookie管理器的使用

测试人生路

接口 Cookie

2021最新京东、字节跳动「3面面经」盘点大厂后端面试高频题

Java架构之路

Java 程序员 架构 面试 编程语言

#滴滴夜莺# Nightingale & Prometheus

漂洋散人

程序员成长第十七篇:项目转测

石云升

项目管理 程序员 28天写作 3月日更

打破垄断 上海发布自主研发树图区块链重大成果

CECBC

区块链

浅谈基于ARP协议的网络攻击

行者AI

网络安全

Vim,人类史上最好用的文本编辑器

沉默王二

vim 开发工具 vim教程

「两次遍历」要比「一次遍历」要慢 ... 为啥呀?为啥呀??

宫水三叶的刷题日记

面试 LeetCode 数据结构与算法

SuperBenchmarker sb在mac上的安装手记

edd

前端上手Docker超详细基础教程

1024肥宅

Docker Linux 大前端 jenkins CI/CD

Serverless 如何在阿里巴巴实现规模化落地?

阿里巴巴云原生

阿里巴巴 Serverless 容器 微服务 云原生

区块链农产品溯源平台,农产品区块链防伪

13530558032

第四章作业(二)

LouisN

七种分布式事务的解决方案,一次讲给你听

moon聊技术

从0到1建立软件测试质量体系

程序员阿沐

软件测试 测试工程师 质量保证

环信大学 | 构建一套适合微服务的高可用架构

DT极客

想做技术自媒体实现财富自由?先看看广告报价吧,无编码学爬虫之二。

梦想橡皮擦

Python 28天写作 2月春节不断更 3月日更

数据库周刊59丨GaussDB(for openGauss)开放商用;人大金仓保障冬奥会演练顺利完成;MDL锁导致的MySQL问题分析;PG日志使用手册;达梦表空间查询;数据库笔试题面试题集……

墨天轮

MySQL 数据库 程序员 运维 postgre

真实字节二面:什么是伪共享?

艾小仙

Java 程序员 字节跳动 面试

2021备战金三银四血拼一波算法:字节+百度+美团+网易+拼夕夕+腾讯+滴滴

比伯

Java 编程 程序员 架构 面试

算力挖矿系统开发|算力挖矿软件APP开发

系统开发

Kubernetes 稳定性保障手册 -- 极简版

阿里巴巴云原生

云计算 容器 开发者 云原生 k8s

Pgbouncer最佳实践:系列一

PostgreSQLChina

数据库 postgresql 软件 开源社区

#集赞送好礼#百度大脑AI开放平台的2020年

百度大脑

四面美团开发岗,成功斩获offer,分享个人面经

Java架构之路

Java 程序员 架构 面试 编程语言

图解CyclicBarrier运动员接力赛

叫练

面试 AQS CyclicBarrier

别梦依稀咒逝川,Ruby二十八年前|M1芯片Mac os系统配置Ruby(3.0.0) on Rails(6.1.1)开发环境(2021最新攻略)

刘悦的技术博客

ruby ruby-on-rails rubygems macOS Big Sur m1

技术解析 | Doris Compaction机制解析

百度开发者中心

百度 apache doris

可变的热信号 RACSubject_语言 & 开发_Draveness_InfoQ精选文章