RxSwift - Observable 阅读笔记

什么是 Observable

在 RxSwift 中，所有的事件都是基于队列 (sequence) 的，Observable 会不断在这条队列上发送数据 / 数据序列。而我们通过订阅 Observable ，当有事件被发送，就可以接受到事件通知。

Observable

Observable 生命周期

Observable 会发送三种事件类型，发送包含事件 element 的 next 事件，发送 completed 事件，作为事件序列的完结。

Completed

当有错误发生时，Observable 会发送包含错误信息的 error(Error) 事件。

Error

Observable 可以一直发送 next 事件，直到发送了 error 或者 completed 事件，这条事件序列才会被终止。

Observable 的创建

RxSwift 提供了一系列创建 Observable 的方法

• just

  创建一个只包含一个事件的序列。

  Observable<Int>.just(1)
  

• of

  创建一个事件序列，包含参数中传递的事件

  Observable.of(1, 2, 3)
  

• from

  通过 sequence 创建事件序列。

  Observable.from([1, 2, 3])
  

• never

  创建一个不会传递任何事件也不会终止的序列

   Observable<Void>.never()
  

• empty

  创建一个只发送 completed 事件的序列

  Observable<String>.empty()
  

• error

  创建一个只发送 error 事件的序列

  Observable<String>.error(MyError.error)
  

• range

  创建一个在某个范围内的 int 值的序列

  Observable<Int>.range(start: 1, count: 10)
  

订阅 Observable

RxSwift 使用 subscribe() 来订阅 Observable。这里需要强调的是，任何 Observable 只有在有订阅者时才会发送事件，因此，例如上文提及的 Observable.of(1, 2, 3) 其实没有发送任何事件。只有当通过 subscribe() 订阅时，Observable 才会发送事件

Observable<Int>
    .of(1, 2, 3)
    .subscribe({ event in
        print(event)
    })

打开控制台，就会得到下面的打印结果

next(1)
next(2)
next(3)
completed

定义的 Observable 对象为每个元素发送了 .next 事件。最后发送了 .completed 事件，作为事件序列的结束。

.next 事件的元素是一个 Optional ，因此当需要直接访问这个值时，需要做解包操作

不过 RxSwift 提供了另一种订阅函数来满足订阅要求。

Observable<Int>
    .of(1, 2, 3)
    .subscribe(onNext: { (element) in
        print(element)
    }, onError: { (error) in
        print(error)
    }, onCompleted: {
        print("Completed")
    }, onDisposed: {
        print("Disposed")
    })

Observable 处理与终止

当 Observable 有订阅者时，才会开始不断发送 .next 事件，直到发送一个 .error 或者 .completed 事件，事件序列结束

我们可以手动的通过取消订阅来终止这个事件序列。

• 单独的为每一个 Observable 添加取消订阅操作

  Observable
      .of(1, 2, 3)
      .subscribe(onNext: { (element) in
          print(element)
      }, onCompleted: {
          print("completed")
      }, onDisposed: {
          print("Disposed")
      })
      .dispose()
  

  控制台的打印结果变成了

  1
  2
  3
  completed
  Disposed
  

  在调用了 dispose() 方法之后，当前的 Observable 对象就停止向订阅者发送消息了。

• 统一管理 Observable 的释放

  为每一个 Observable 都添加 dispose() 方法不仅显得多余，并且如果在多个地方使用到这个 Observable 对象，在某处提前释放了，可能会导致一些意想不到的问题。RxSwift 提供了一个 DisposeBag 来统一管理 Observable 的释放问题

  let bag = DisposeBag()
  Observable
      .of(1, 2, 3)
      .subscribe(onNext: { (element) in
          print(element)
      })
      .addDisposableTo(bag)
  

  首先定义一个 DisposeBag 对象，之后就像使用 dispose() 一样，在订阅之后使用 addDisposableTo(_ bag: RxSwift.DisposeBag) 将 Observable 添加到定义的 DisposeBag 对象中。

  DisposeBag 将是后面经常使用到的处理 Observable 回收的方法。

自定义事件

循规蹈矩的 Observable 的创建方式，或多或少都无法满足日常 PM 的需求。所以如何创建包含自定义事件的数据流呢？RxSwift 提供了两种方式

• create

  let disposeBag = DisposeBag()
  enum MyError: Error {
      case anError
  }
  Observable<String>
      .create({ (observer) in
          observer.onNext("1")
          observer.onError(MyError.anError)
          observer.onCompleted()
          observer.onNext("where is it?")
          return Disposables.create()
      })
  

  create 函数定义

  public static func create(_ subscribe: @escaping (RxSwift.AnyObserver<RxSwift.Observable.E>) -> Disposable) -> RxSwift.Observable<RxSwift.Observable.E>

  在传递的闭包内部，我们可以自定义事件，对上述的 Observable 订阅，我们将接收到下列事件

  1
  anError
  Disposed
  

  在接收到 .error 事件之后，整个事件序列就被终止了。

• deferred 创建一个生成不同的事件序列的工厂方法

  试想这样的使用场景，对于同一个 Observable，我们希望不同的订阅者能订阅到不同的数据流。

  let disposeBag = DisposeBag()
  var flip = false
  let factory: Observable<Int> = Observable.deferred({
      flip = !flip
      
      if flip {
          return Observable.of(1, 2, 3)
      } else {
          return Observable.of(4, 5, 6)
      }
  })
  for _ in 0 ... 3 {
      factory.subscribe(onNext: {
          print($0, terminator: "")
      })
      .disposed(by: disposeBag)
      print()
  }
  

  每个订阅 factory 的订阅者，都能获得不同的事件序列，控制台的打印结果为

  123
  456
  123
  456