Observable介绍

Observable<T>类是RxSwift框架的基础。其作用就像是一条流水线，让观察者可以实时获取对所有可观察对象所触发的事件，也就是说以此来实现对UI的实时更新或数据、事件等实时处理。
与之前介绍RAC类似，Observable对象所触发的事件有：
next，触发时将可观察对象的新值传递给观察者
completed，可观察对象的生命周期正常结束并不再响应触发事件
error，可观察对象出现错误导致其生命周期终止

贴一个最基本的Observable的创建与订阅代码

        //创建Observable
        let ob = Observable<Any>.create { (anyObser) -> Disposable in
            print("this is a observable create")
            print(anyObser)
            
            anyObser.onNext("执行下一步")
            anyObser.onCompleted()
            anyObser.onNext("执行第二步")
            anyObser.onError(testError.errorB)
            
            return Disposables.create()
        }
        
        //订阅Observable
        ob.subscribe(onNext: { (signal) in
            print("next === \(signal)")
        }, onError: { (error) in
            print("error == \(error)")
        }, onCompleted: {
            print("完成订阅")
        }) {
            print("销毁订阅")
        }.disposed(by: disposeBag)  //disposeBag为创建的垃圾包

接触过RAC系列的，会发现RxSwift语法结构非常相似。此处特意把error事件放在completed事件之后，打印结果证明，观察者在触发complete事件之后不会再响应任何事件。

打印结果

Observable创建

以下列出Observable各种创建的方法，以及对应不同的使用场景

just方法

该方法通过传入一个默认值来初始化

        let createJustObservable = Observable.just("one")
        createJustObservable
.subscribe { (event) in
            print(event)
        }.disposed(by: disposeBag)

//为了方便查看，将上述代码合并
        Observable.just("one")
            .subscribe { (event) in
            print(event)
        }.disposed(by: disposeB)

of方法

该方法可以接受可变数量的参数传入，但必需时同类型的

        Observable.of("o","f","of").subscribe { (event) in
            print(event)
            }.disposed(by: disposeBag)

from方法

该方法需要一个数组参数传入

        Observable.from(["f","r","o","m"]).subscribe { (event) in
            print(event)
            }.disposed(by: disposeBag)

never方法

该方法创建一个永远不会发出 Event（也不会终止）的 Observable 序列

        Observable<Int>.never()
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

empty方法

该方法创建一个空内容的 Observable 序列，执行时直接执行completed方法

        Observable<Int>.empty()
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

error方法

该方法创建一个不做任何操作，而是直接发送一个错误的 Observable 序列

//首先创建一个error类型的枚举
enum MyError:Error {
    case A
    case B
    var errorType:String {
        switch self {
        case .A:
            return "i am error A"
        case .B:
            return "BBBB"
        }
    }
}

        let myError = MyError.A
        //        print(myError.errorType)
        Observable<Int>.error(myError)
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

range方法

该方法通过指定起始和结束数值，创建一个以这个范围内所有值作为初始值的Observable序列

        Observable
            .range(start: 1, count: 6)
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

repeatElement方法

该方法创建一个可以无限发出给定元素的 Event的 Observable 序列（永不终止）

            Observable.repeatElement("repeat")
                .subscribe { (event) in
                print(event)
            }
                .disposed(by: disposeBag)

generate方法

该方法创建一个只有当提供的所有的判断条件都为 true 的时候，才会给出动作的 Observable 序列

        //第一个参数：初始化的数值为0
        //第二个参数：判断条件闭包第一个元素$0是否小于等于10
        //第三个参数：使$0的int值+2
        //若判断条件为true，依次循环下去
        Observable
            .generate(initialState: 0, condition: {$0<=10}, iterate: {$0+2})
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

上面的方法也可以用of方法表示，但灵活性不足

        Observable.of(0,2,4,6,8,10)
            .subscribe { (event) in
            print(event)
            }
            .disposed(by: disposeBag)

deferred方法

该方法相当于是创建一个 Observable 工厂，通过传入一个 block 来执行延迟 Observable序列创建的行为，而这个 block 里才是真正的实例化Observable序列对象

        //此处也可将isOdd属性修改为false来查看结果
        var isOdd = true
        let factory: Observable<Int> = Observable
            .deferred { () -> Observable<Int> in
            
            isOdd = !isOdd
            if isOdd{
                return Observable.of(0,2,4,6,8)
            }else{
                return Observable.of(1,3,5,7,9)
            }
        }
        
        factory.subscribe { (event) in
            print("\(isOdd)",event)
            }
            .disposed(by: disposeBag)

interval和timer方法

这两个方法都是创建的 Observable 序列每隔一段设定的时间，会发出一个索引数的元素，而且会一直发送

        Observable<Int>
            .timer(1, scheduler: MainScheduler.instance).subscribe{(event) in
            print("scheduler",event)
        }
            .disposed(by: disposeBag)


        Observable<Int>
            .timer(5, period: 1, scheduler: MainScheduler.instance).subscribe { (event) in
            print("timer",event)
            }
            .disposed(by: disposeBag)

Observable订阅与销毁

与 RAC 的订阅信号方法非常类似，使用过程中是需要在需要订阅 Observable 的地方调用 subscribe 方法即可。在RxSwift中每一个订阅都是唯一的，而且没有一个类似NotificationCenter通知机制 default 这样的全局单例对象。当没有订阅者时，Observable 对象不会发送通知。
在创建Observable时，在订阅任何不同的观察者之后，代码一定会添加一行 .disposed(by: disposeBag) 代码，而 disposeBag 是之前全局创建生成的let disposeBag = DisposeBag()。此处代码是将所有的订阅者通过RxSwift提供的集中销毁管理垃圾包来集中销毁订阅信号。若不这么做，Observable 对象在生命周期完结时会存在内存泄漏的问题引发崩溃。

该文章首次发表在 简书：我只不过是出来写写代码 博客，并自动同步至 腾讯云：我只不过是出来写写iOS 博客