RxSwift核心逻辑

一、Observable
二、Observer
三、Subject
四、Operator
五、Disposable
六、Schedulers
七、Error Handling

凡事物皆序列

一、Observable

Observable为可监听序列，用于描述一个对象所生产的序列，任何序列都可以用Observable来标识。

sequence.png

万事万物都为序列，每个个体都是观察者，观察周围事物的变化。

页面开发中的序列

  • scrollview页面滑动
    开发中，会遇到scrollview滑动监听位置的需求，在某某位置，修改导航透明度。那么位置变化所产生的一系列的位置参数就构成了一个序列。如下图：
scroll.png
  • 定时器序列
    用户注册信息，为验证用户信息，需要触发获取验证码按钮，一般60秒内只能获取一次短信，因此在发送请求后60秒内，按钮应该处于失能状态，这时就需要一个60秒倒计时定时器，60秒后恢复按钮。如下图：
time.png

以上都是对某一个点的监听，除以上还有button点击，还有textField的编辑，手势的监听，switch的状态等等这些事件都可以抽象为对序列的监听。通常对于监听有着对应的触发方法，或代理。在RxSwift中呢，则是对这些事件做了封装，通过observer来监听事件的到来。

Observable是对这些不同类型序列的统一管理，看一下具体是一个什么类：

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
    ……
}

是一个继承自ObservableType的一个类，先不管ObservableType这个类，重点是Observable是一个泛型类，好像是明白了些什么，泛型就是不指定类型，传入什么类型，Observable就去找对应类型，如button，textview等等对应的一些序列，方法公用，统一管理各种类型序列。

序列可表示的不同类型如下：

Observable<Int>
Observable<String>
Observable<JSON>
Observable<Void>
……

创建一个序列并监听序列消息发送：

//1、创建一个序列
let ob = Observable<Int>.create { observer in
    //3、发送一个消息
    observer.onNext(1)
    observer.onNext(2)
    return Disposables.create()
}
//2、监听序列消息
ob.subscribe(onNext: { (val) in
    print(val)
}).disposed(by: disposeBag)

对于Observable还存在一些特征序列就不一一介绍。RxSwift核心逻辑

二、Observer

Observer观察者，观察一系列事件的产生，根据需要做出必要的操作，如同button的点击事件，textfield的代理方法等等，实现后便是一个观察者，而在Rx中则被抽象出来为一个观察者Observer。如下图：

observer.png

下面看一下观察者是怎么抽象出来的，如下代码：

KVO：

self.person.rx.observeWeakly(String.self, "name").subscribe(onNext: { (value) in
    print(value as Any)
}).disposed(by: disposeBag)

UIButton：

button.rx.tap.subscribe(onNext: { () in
    print("被点击了")
    //处理事件
}).disposed(by: disposeBag)

UITextField：

textfield.rx.text.orEmpty.changed.subscribe(onNext: { (text) in
    print(text)
}).disposed(by: disposeBag)

通过以上代码观察到，KVO的通知，按钮的点击及文本框的输入，均回调到subscribe的闭包中（参数闭包）。点击进入这个事件（可以通过cmd+点击进入查看）可以找到如下：

public enum Event<Element> {
    /// Next element is produced.
    case next(Element)
    /// Sequence terminated with an error.
    case error(Swift.Error)
    /// Sequence completed successfully.
    case completed
}
类型 说明
next 是产生一个新的元素，参数为任意类型
error 创建序列所产生的错误，如网络请求失败，终止序列
completed 所有序列都已经发送完成

是一个能够接收任意类型的枚举，说明是所有事件回调的媒介。
这里的观察者和订阅写在一起，下面再看看其他的写法，AnyObserver、Binder。

1、AnyObserver

表示任意观察者。如网络请求：

let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .subscribe(onNext: { (data) in
        print(data)
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成")
    }).disposed(by: disposeBag)

等价于：

let observer: AnyObserver<(response: HTTPURLResponse, data: Data)> = AnyObserver { (event) in
    switch event {
    case .next(let data):
        print(data)
    case .error(let error):
        print("Data Task Error: \(error)")
    default:
        break
    }
}
let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .subscribe(observer).disposed(by: disposeBag)

将观察者和订阅方法分离。除此之外所有的观察者都可以被抽离。

2、Binder

有两个特征：

  • 不会处理错误事件
  • 确保绑定都是在给定Scheduler上执行（默认在MainScheduler）

主要在UI绑定上使用，如textfield绑定到label，由输入产生时间，根据设定条件设置显隐。

文本框验证：

let passwordValid = pwdTf.rx.text.orEmpty
            .map{$0.count>6}
            .share(replay: 1, scope: .whileConnected)
passwordValid.bind(to: pwdLabel.rx.isHidden).disposed(by: disposeBag)

内部设置如下：

extension Reactive where Base: UIView {
    public var isHidden: Binder<Bool> {
        return Binder(self.base) { view, hidden in
            view.isHidden = hidden
        }
    }
}

所有继承自UIView子控件的绑定显隐都通过该扩展实现的。

按钮点击绑定：

//联合绑定
Observable.combineLatest(nameValid,pwdValid){
    $0 && $1
}.bind(to: button.rx.isEnabled)
.disposed(by: disposeBag)

内部设置如下：

extension Reactive where Base: UIControl {
    public var isEnabled: Binder<Bool> {
        return Binder(self.base) { control, value in
            control.isEnabled = value
        }
    }
}

label文本绑定：

//监听输入变化
textfield.rx.text.orEmpty.changed.subscribe(onNext: { (text) in
    print(text)
}).disposed(by: disposeBag)
//绑定，数据传递
textfield.rx.text.bind(to: label.rx.text)

内部设置如下：

extension Reactive where Base: UILabel {
    public var text: Binder<String?> {
        return Binder(self.base) { label, text in
            label.text = text
        }
    }
}

通过以上方法，我们也可以扩展方法来实现一个自定义的观察者。

三、Subject

Observable & Observer即是可监听序列也是观察者，如文本框输入监听，text即作为序列输出，text绑定作为观察者。switch、slider、segment等都具有此特性，观察值的变化并给被绑定对象的text赋值。此外框架还定义了其他辅助类，可以帮助开发者准确的描述事务特征。

1、AsyncSubject

使用结合.onCompleted()来使用，无论有多少序列（每个元素是一个事件），只发送由源产生的最后一个事件。事件由onCompleted和onError的调用而终止，否则发送的方法不会触发，不会再产生新的事件。代码如下：

let sub = AsyncSubject<Any>()
sub.subscribe(onNext: { (val) in
    print(val)
}, onCompleted: {
    print("完成")
}).disposed(by: disposeBag)
sub.onNext("1")
sub.onCompleted()
sub.onNext("2")
sub.onCompleted()

打印：

value:1
完成

完成后订阅立即取消，后边的消息发送均无观察者接收。

2、PublishSubject

只有先订阅成为观察者，才能收到发送的消息，多次订阅，一个消息会发送给多个观察者，如发送onCompleted和onError将终止所有事件的产生和发送。代码如下：

let sub = PublishSubject<Any>()
sub.subscribe{
    print("channal1:\($0)")
    }.disposed(by: disposeBag)
sub.onNext("hibo")
sub.onNext("luck")

sub.subscribe{
    print("channal2:\($0)")
    }.disposed(by: disposeBag)

sub.onNext("coder")
sub.onNext("sony")

打印：

channal1:next(hibo)
channal1:next(luck)
channal1:next(coder)
channal2:next(coder)
channal1:next(sony)
channal2:next(sony)

3、ReplaySubject

无论何时订阅ReplaySubject，存储最后n次onNext发送出的元素，由bufferSize决定存储新值的个数。当有新的订阅者订阅时直接发送存储的最新的元素。代码如下：

let sub = ReplaySubject<Any>.create(bufferSize: 1)
sub.subscribe{
    print("channal1:\($0)")
}.disposed(by: disposeBag)
sub.onNext("hibo")
sub.onNext("luck")

sub.subscribe{
    print("channal2:\($0)")
}.disposed(by: disposeBag)

sub.onNext("coder")
sub.onNext("sony")

打印：

channal1:next(hibo)
channal1:next(luck)
channal2:next(luck)  此处为订阅后立即收到的消息
channal1:next(coder)
channal2:next(coder)
channal1:next(sony)
channal2:next(sony)

4、BehaviorSubject

通过一个初始值来创建，每次订阅BehaviorSubject后，会接收到subject发出的Event，传递值为初始值。代码如下：

let sub = BehaviorSubject(value: "init_value")
sub.subscribe{
    print("channal1:\($0)")
}.disposed(by: disposeBag)
sub.onNext("hibo")
sub.onNext("luck")

sub.subscribe{
    print("channal2:\($0)")
}.disposed(by: disposeBag)

sub.onNext(“coder")
sub.onNext("sony")

打印：

channal1:next(init_value)  订阅后即发送的消息
channal1:next(hibo)        调用onNext发送
channal1:next(luck)           
channal2:next(luck)        订阅后将发送最后一次发送的消息
channal1:next(coder)
channal2:next(coder)
channal1:next(sony)
channal2:next(sony)

5、ControlProperty

用于描述UI控件属性。具有以下特征：

  • 不会产生error事件
  • 在主线程（MainScheduler）订阅
  • 在主线程监听
  • 共享状态变化

最典型的代码实列及，文本框输入绑定，textfield触发修改label的text值。如下：

//监听输入变化
textfield.rx.text.orEmpty.changed.subscribe(onNext: { (text) in
    print(text)
}).disposed(by: disposeBag)
//绑定，数据传递
textfield.rx.text.bind(to: label.rx.text)

四、Operator

操作符，可以产生新的序列，或改变组合原有序列生成新序列，也就是提供了额外处理序列的功能，使得功能性更强。如以上所提的.map、.combineLatest等操作符，完成对序列的筛选，和验证的联合绑定。如下代码：

let nameValid = textfield.rx.text.orEmpty
    .map{$0.count>3}
    .share(replay: 1, scope: .whileConnected)
nameValid.bind(to: nameLabel.rx.isHidden).disposed(by: disposeBag)

let pwdValid = pwdTf.rx.text.orEmpty
    .map{$0.count>6}
    .share(replay: 1, scope: .whileConnected)
pwdValid.bind(to: pwdLabel.rx.isHidden).disposed(by: disposeBag)

//联合绑定-需要序列成对出现
Observable.combineLatest(nameValid,pwdValid){
    $0 && $1
}.bind(to: button.rx.isEnabled).disposed(by: disposeBag)

除此之外还很很多类似功能的操作符，该操作符是对ObservableType的扩展，这里只做简单介绍，可以点击进入查看。

五、Disposable

可被清除的资源。用来取消订阅，并清除订阅过程中的内部资源。清除方法直接调用dispose()方法订阅将被取消，再发送消息，该观察者已不存在。发送onCompleted和onError消息，订阅也会被取消，并清除内部资源。

let sub = PublishSubject<Any>()
let dispose = sub.subscribe(onNext: { (val) in
    print("value:\(val)")
})
sub.onNext("11111")
dispose.dispose()  //此处调用onCompleted、onError作用相同
sub.onNext("22222")

打印：11111

DisposeBag 清除包取消订阅，清除内部资源。与dispose不同之处，dispose是手动清除调用即清除，DisposeBag对所有订阅者做打包处理，在当前视图被释放，DisposeBag也被释放，其管理的订阅者同时都被释放掉。所以使用DisposeBag即可，在每一个订阅者后调用该方法，就不用考虑什么时候需要取消订阅。当然如果能够准确控制订阅的取消，更好。如下代码：

let sub = PublishSubject<Any>()
sub.subscribe(onNext: { (val) in
    print("value:\(val)")
}).disposed(by: disposeBag)
sub.onNext("11111")
sub.onNext("22222")

如果知道什么时候结束我们可以直接调用onCompleted方法而disposed就不需要了。

六、Schedulers

调度器是Rx的多线程核心模块，决定任务在哪个线程的哪个队列中执行。常规对GCD的使用如下：

DispatchQueue.global(qos: .userInitiated).async {
    var total = 0
    for index in 1...10{
        total += index
    }
    DispatchQueue.main.async {
        print(total)
    }
}

在全局队列中对一组数据做异步运算，结束之后返回主队列打印。Rx对GCD的实现：

let obj = Observable<Int>.create { (oberver) -> Disposable in
    var total = 0
    for index in 1...10{
        total += index
    }
    oberver.onNext(total)
    return Disposables.create()
}
obj .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated))//选择一个并行队列
    .observeOn(MainScheduler.instance)//设置观察者在主线程上
    .subscribe(onNext: { (val) in
        print(val)
    }).disposed(by: disposeBag)

这样编码，感觉代码量变的很大，但是Rx让代码结构变的更清晰，并且统一了对事件的观察，统一思想快速开发。

线程在Rx中的表示线程的类：

类型 说明
MainScheduler 主线程类
SerialDispatchQueueScheduler 串行队列类
ConcurrentDispatchQueueScheduler 并行队列类
OperationQueueScheduler 具备NSOperationQueue的特点

七、Error Handling

错误处理。序列中也是数据处理，会有逻辑错误。RxSwift提供了两种错误处理机制

  • retry 重试
  • catch 恢复

retry 示例如下：

let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .retry(3)
    .subscribe(onNext: { (data) in
        print(data)
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成")
    }).disposed(by: disposeBag)

断开网络测试打印显示发送3条请求，信息太长这里就不粘贴输出结果了。如果再请求失败后需要等待几秒种在做请求可以使用retryWhen，示例：

let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .retryWhen{ (error: Observable<Error>) -> Observable<Int> in
        return Observable.timer(3, scheduler: MainScheduler.instance)
    }
    .subscribe(onNext: { (data) in
        print(data)
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成")
    }).disposed(by: disposeBag)

运行观察，失败后5秒重新发起一次请求。还可以设置请求次数，当然不是结合.retry使用，示例如下：

let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .retryWhen{ (error: Observable<Error>) -> Observable<Int> in
        return error.enumerated().flatMap({ (index,error) -> Observable<Int> in
            guard index<3 else{
                return Observable.error(error)
            }
            return Observable<Int>.timer(3, scheduler: MainScheduler.instance)
        })
    }
    .subscribe(onNext: { (data) in
        print(data)
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成")
    }).disposed(by: disposeBag)

catch 示例如下：

let url = URL.init(string: "https://www.baidu.com")
//预处理信息
let data = Observable<(response: HTTPURLResponse, data: Data)>.create { (observer) -> Disposable in
    observer.onNext((HTTPURLResponse.init(),Data.init()))
    return Disposables.create()
}
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .catchError({ (error) -> Observable<(response: HTTPURLResponse, data: Data)> in
        //返回预处理数据，防止执行onError 返回的数据走到 onNext下
        return data
    })
    .subscribe(onNext: { (data) in
        //成功失败都会走到此处
        print("data:\(data)")
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成")
    }).disposed(by: disposeBag)

可以使得错误信息发送至onNext下。如果需要一个备用值可使用catchErrorJustReturn，示例如下：

URLSession.shared.rx.response(request: URLRequest.init(url: url!))
    .catchErrorJustReturn((HTTPURLResponse.init(), Data.init()))
    .subscribe(onNext: { (data) in
        //成功失败都会走到此处
        print("data:\(data)")
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("请求完成”) //会执行onCompleted
    }).disposed(by: disposeBag)

通过以上对Rx的了解，总结如图：

RxSwift.png

参考：RxSwift中文文档

下一节 RxSwift核心源码探索

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,684评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,143评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,214评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,788评论 1赞 277
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,796评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,665评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,027评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,679评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 41,346评论 1赞 299
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,664评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,766评论 1赞 331
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,412评论 4赞 321
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,015评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,974评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,203评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,073评论 2赞 350
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,501评论 2赞 343