RxSwift2-创建、订阅、销毁Observable

2021-04-13 本文已影响0人浮华_du

本篇文章,介绍下创建Observable的几种方式,以及如何订阅销毁Observable

创建Observable

1.just()

传入一个默认值
指定序列泛型

        let observable0 = Observable<Int>.just(5)
       //订阅测试
        observable0.subscribe {
            print("just()_\($0)")
        }.disposed(by: disposeBag)
// just()_next(5)
// just()_completed

2.of()

传入可变数量的默认值(类型一致)
指定序列泛型(可不指定,会自动推断)

        let observable1 = Observable<String>.of("A", "B", "C")
        //订阅测试
       observable1.subscribe {
            print("of()_\($0)")
        }.disposed(by: disposeBag)

//of()_next(A)
//of()_next(B)
//of()_next(C)
//of()_completed

3.from()

指定一个数组参数

        let observable2 = Observable.from(["A", "B", "C"])
        //订阅测试
      observable2.subscribe {
            print("from()_\($0)")
        }.disposed(by: disposeBag)

//from()_next(A)
//from()_next(B)
//from()_next(C)
//from()_completed

4.empty()

创建一个空内容的 Observable 序列

        let observable3 = Observable<Int>.empty()
        //订阅测试
      observable3.subscribe {
            print("empty()_\($0)")
        }.disposed(by: disposeBag)

//empty()_completed

5.never()

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

        let observable4 = Observable<Int>.never()
        //订阅测试
        observable4.subscribe {
            print("never()_\($0)")
        }.disposed(by: disposeBag)

6.error()

直接发送一个错误的 Observable 序列

enum MyError: Error {
        case A
        case B
    }
        let observable5 = Observable<Int>.error(MyError.A)
       //订阅测试
       observable5.subscribe {
            print("error()_\($0)")
        }.disposed(by: disposeBag)

//error()_error(A)

7.range()

指定起始值和结束值,生成从起始值到结束值的序列

        let observable6 = Observable.range(start: 1, count: 5)
       //订阅测试
      observable6.subscribe {
            print("range()_\($0)")
        }.disposed(by: disposeBag)

//range()_next(1)
//range()_next(2)
//range()_next(3)
//range()_next(4)
//range()_next(5)
//range()_completed

8.repeatElement()

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

        let observable7 = Observable.repeatElement(1)
       observable7.subscribe {//订阅之后会阻塞UI
               print("repeatElement()_\($0)")
       }.disposed(by: disposeBag)

///repeatElement()_next(1)
///repeatElement()_next(1)
///repeatElement()_next(1)
///...

9.generate()

该方法创建一个只有当提供的所有的判断条件都为 true 的时候，才会给出动作的 Observable 序列。(需要传入: 初始值、迭代、判断条件)

 let observable8 = Observable.generate(
            initialState: 0,
            condition: { $0 <= 10 },
            iterate: { $0 + 2 }
        )

observable8.subscribe {
            print("generate()_\($0)")
        }.disposed(by: disposeBag)

//generate()_next(0)
//generate()_next(2)
//generate()_next(4)
//generate()_next(6)
//generate()_next(8)
//generate()_next(10)
//generate()_completed

10.create()

该方法接受一个 block 形式的参数，任务是对每一个过来的订阅者进行处理


        //这个block有一个回调参数observer就是订阅这个Observable对象的订阅者
        //当一个订阅者订阅这个Observable对象的时候，就会将订阅者作为参数传入这个block来执行一些内容
        let observable9 = Observable<String>.create{ observer in
            //对订阅者发出了.next事件，且携带了一个数据"啊哈哈"
            observer.onNext("啊哈哈")
            //对订阅者发出了.completed事件
            observer.onCompleted()
            //因为一个订阅行为会有一个Disposable类型的返回值，所以在结尾一定要returen一个Disposable
            return Disposables.create()
        }

observable9.subscribe {
            print("create()_\($0)")
        }.disposed(by: disposeBag)

//create()_next(啊哈哈)
//create()_completed

11.deferred()

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

//用于标记是奇数、还是偶数
        var isOdd = true
        
        //使用deferred()方法延迟Observable序列的初始化，通过传入的block来实现Observable序列的初始化并且返回。
        let factory : Observable<Int> = Observable.deferred {
            
            //让每次执行这个block的时候都会让奇、偶数进行交替
            isOdd = !isOdd
            
            //根据isOdd参数，决定创建并返回的是奇数Observable、还是偶数Observable
            if isOdd {
                return Observable.of(1, 3, 5 ,7)
            }else {
                return Observable.of(2, 4, 6, 8)
            }
        }

         //第1次订阅测试
        factory.subscribe { event in
            print("factory__deferred()__\(isOdd)", event)
        }.disposed(by: disposeBag)
        
        //第2次订阅测试
        factory.subscribe { event in
            print("factory__deferred()__\(isOdd)", event)
        }.disposed(by: disposeBag)

//factory__deferred()__false next(2)
//factory__deferred()__false next(4)
//factory__deferred()__false next(6)
//factory__deferred()__false next(8)
//factory__deferred()__false completed
//factory__deferred()__true next(1)
//factory__deferred()__true next(3)
//factory__deferred()__true next(5)
//factory__deferred()__true next(7)
//factory__deferred()__true completed

12.interval()

创建的 Observable 序列每隔一段设定的时间，会发出一个索引数的元素。而且它会一直发送下去。
(下面方法让其每 1 秒发送一次，并且是在主线程（MainScheduler）发送。)

        let observable11 = Observable<Int>.interval(1, scheduler: MainScheduler.instance)

observable11.subscribe { event in
            print("interval()__\(event)")
        }.disposed(by: disposeBag)
//interval()__next(0)
//interval()__next(1)
//interval()__next(2)
//interval()__next(3)
//interval()__next(4)
// ... 每一秒打印一次, 会一直执行下去

///为了防止会一直执行下去,我们在5秒后,将这个RxSwift.Disposable dispose
let subInterval = observable11.subscribe { event in
            print("interval()__\(event)")
   }
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 5.0) {
            subInterval.dispose()
}
//interval()__next(0)
//interval()__next(1)
//interval()__next(2)
//interval()__next(3)
//interval()__next(4)

13.timer()

创建的 Observable序列在经过设定的一段时间后，产生唯一的一个元素。

      ///5秒后发送一次
        let observable12 = Observable<Int>.timer(1, scheduler: MainScheduler.instance)

        observable12.subscribe { event in
            print("timer()__\(event)")
        }.disposed(by: disposeBag)
//timer()__next(0)
//timer()__completed

创建的 Observable 序列在经过设定的一段时间后，每隔一段时间产生一个元素。

        /// 5秒后,每一秒产生一个元素
        let observable13 = Observable<Int>.timer(5, period: 1, scheduler: MainScheduler.instance)

observable13.subscribe(onNext: { (event) in
            print("timer()__\(event)")
        }, onError: { (error) in
            print("timer()__\(error)")
        }, onCompleted: {
            print("timer()__onCompleted")
        }) {
            print("timer()__onDisposed")
        }

//timer()__0
//timer()__1
//timer()__2
//timer()__3
//timer()__4
//timer()__5
//timer()__6
//timer()__7
//timer()__8
//timer()__9
//...


//为了防止会一直执行下去,我们在10秒后,将这个RxSwift.Disposable dispose
let subscription = observable13.subscribe(onNext: { (event) in
            print("timer()__\(event)")
        }, onError: { (error) in
            print("timer()__\(error)")
        }, onCompleted: {
            print("timer()__onCompleted")
        }) {
            print("timer()__onDisposed")
        }

 DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 10.0) {
            subscription.dispose()
       }
//timer()__0
//timer()__1
//timer()__2
//timer()__3
//timer()__4
//timer()__5
//timer()__onDisposed

以上,就是创建Observable的各种方式.

subscribe --订阅Observable

通过上述测试可知, 我们都是使用 subscribe() 方法来订阅Observable，接收它发出的 Event.

使用 subscribe() 订阅了一个Observable 对象，该方法的 block 的回调参数就是被发出的 event 事件

初始化 Observable 序列时设置的默认值都按顺序通过 .next 事件发送出来。
当 Observable 序列的初始数据都发送完毕，它还会自动发一个 .completed 事件出来。

        let observable0 = Observable<Int>.just(5)
        observable0.subscribe {
            print("just()_\($0)")
        }

//just()_next(5)
//just()_completed

2.使用 subscribe() 订阅了一个Observable 对象,通过不同的 block 回调处理不同类型的 event.(onNext、onError、onCompleted 和 onDisposed 这四个回调 block 参数都是有默认值的，即它们都是可选的。所以我们也可以只处理 onNext而不管其他的情况)

        let observable0 = Observable<Int>.just(5)
        observable0.subscribe { (event) in
            print("just()__\(event)")
        } onError: { (error) in
            print("just()__error")
        } onCompleted: {
            print("just()__onCompleted")
        } onDisposed: {
            print("just()__onDisposed")
        }
//just()__5
//just()__onCompleted
//just()__onDisposed

监听事件的生命周期 doOn()

我们可以使用 doOn 方法来监听事件的生命周期，它会在每一次事件发送前被调用。
同时它和 subscribe 一样，可以通过不同的block 回调处理不同类型的 event
通过例子可以看出,

doOn的onNext onCompleted比subscribe的onNext onCompleted先执行;
subscribe的onDisposed比doOn的onDisposed先执行

        let observable1 = Observable<String>.of("A", "B", "C")

observable1.do(onNext: { (ele) in
            print("doOn---\(ele)")
        }, onError: { (error) in
            print("doOn---\(error)")
        }, onCompleted: {
            print("doOn---onCompleted")
        }, onSubscribe: {
            print("doOn---onSubscribe")
        }, onSubscribed: {
            print("doOn---onSubscribed")
        }, onDispose: {
            print("doOn---onDispose")
        }).subscribe { (event) in
            print("subscribe---\(event)")
        } onError: { (error) in
            print("subscribe---\(error)")
        } onCompleted: {
            print("subscribe---onCompleted")
        } onDisposed: {
            print("subscribe---onDisposed")
        }.disposed(by: disposeBag)

//doOn---onSubscribe
//doOn---onSubscribed
//doOn---A
//subscribe---A
//doOn---B
//subscribe---B
//doOn---C
//subscribe---C
//doOn---onCompleted
//subscribe---onCompleted
//subscribe---onDisposed
//doOn---onDispose

Observable 的销毁（Dispose）

一个 Observable 序列被创建出来后它不会马上就开始被激活从而发出 Event，而是要等到它被某个人订阅了才会激活它。
而 Observable 序列激活之后要一直等到它发出了.error或者 .completed的 event 后，它才被终结。

1.dispose()

使用该方法我们可以手动取消一个订阅行为.(例子可参考上面interval() timer() 初始化方式中, 取消订阅的行为)

2.DisposeBag

DisposeBag 的对象来管理多个订阅行为的销毁：
我们可以把一个 DisposeBag对象看成一个垃圾袋，把用过的订阅行为都放进去。
而这个DisposeBag 就会在自己快要dealloc 的时候，对它里面的所有订阅行为都调用 dispose()方法。(上文中其实我们已经在使用这个垃圾袋了哦~)

    private var disposeBag:DisposeBag = DisposeBag()