可觀察的序列Observable

經過前面博客對Rxswift的源碼分析，咱們知道在Rxswift中一條主線思想就是萬物皆序列，這裏的序列就是咱們的可觀察序列，也能夠稱之爲觀察者。因此使用Rxswift總會跟觀察者Observer打交道，這裏重溫一下觀察者的定義。oop

本篇博客主要講解Observer的建立，訂閱，銷燬的用法，不深刻講解它的底層實現，不講解它的源碼，如要了解源碼實現過程分析請參考我以前的博客：序列的核心邏輯

Observable定義

觀察者（Observable）的做用就是監聽事件，而後對這個事件作出響應。或者說任何響應事件的行爲都是觀察者。好比：

當咱們點擊按鈕，彈出一個提示框。那麼這個「彈出一個提示框」就是觀察者Observer
當咱們請求一個遠程的json 數據後，將其打印出來。那麼這個「打印 json 數據」就是觀察者 Observer

Observable的建立，訂閱，銷燬

Observable建立

在 subscribe 方法中建立Observable

建立觀察者最直接的方法就是在 Observable 的 subscribe 方法後面描述當事件發生時，須要如何作出響應。
好比下面的樣例，觀察者就是由後面的 onNext，onError，onCompleted 這些閉包構建出來的。
實例1：

代碼：

let observable = Observable.of("A", "B", "C")
          
observable.subscribe(onNext: { element in
    print(element)
}, onError: { error in
    print(error)
}, onCompleted: {
    print("completed")
})
複製代碼

運行結果：

A
B
C
completed
複製代碼

在 bind 方法中建立Observable

實例2：下面代碼咱們建立一個定時生成索引數的 Observable 序列，並將索引數不斷顯示在 label 標籤上

代碼：

import UIKit
import RxSwift
import RxCocoa
 
class ViewController: UIViewController {
     
    @IBOutlet weak var label: UILabel!
     
    let disposeBag = DisposeBag()
     
    override func viewDidLoad() {
         
        //Observable序列（每隔1秒鐘發出一個索引數）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
 
        observable
            .map { "當前索引數：\($0 )"}
            .bind { [weak self](text) in
                //收到發出的索引數後顯示到label上
                self?.label.text = text
            }
            .disposed(by: disposeBag)
    }
}
複製代碼

使用 AnyObserver 建立Observable

AnyObserver 能夠用來描敘任意一種觀察者。

AnyObserver配合 subscribe 方法使用好比上面實例1咱們能夠改爲以下代碼：

//觀察者
let observer: AnyObserver<String> = AnyObserver { (event) in
    switch event {
    case .next(let data):
        print(data)
    case .error(let error):
        print(error)
    case .completed:
        print("completed")
    }
}
 
let observable = Observable.of("A", "B", "C")
observable.subscribe(observer)
複製代碼

AnyObserver配合bindTo 方法使用也可配合 Observable 的數據綁定方法（bindTo）使用。好比上面的實例2能夠改爲以下代碼：

import UIKit
import RxSwift
import RxCocoa
 
class ViewController: UIViewController {
     
    @IBOutlet weak var label: UILabel!
    let disposeBag = DisposeBag()
    override func viewDidLoad() {
         
        //觀察者
        let observer: AnyObserver<String> = AnyObserver { [weak self] (event) in
            switch event {
            case .next(let text):
                //收到發出的索引數後顯示到label上
                self?.label.text = text
            default:
                break
            }
        }
         
        //Observable序列（每隔1秒鐘發出一個索引數）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        observable
            .map { "當前索引數：\($0 )"}
            .bind(to: observer)
            .disposed(by: disposeBag)
    }
}
複製代碼

使用 Binder 建立Observable

相較於AnyObserver 的大而全，Binder 更專一於特定的場景。Binder 主要有如下兩個特徵：

不會處理錯誤事件
確保綁定都是在給定 Scheduler 上執行（默認 MainScheduler）

一旦產生錯誤事件，在調試環境下將執行 fatalError，在發佈環境下將打印錯誤信息。

針對上面的實例2 中的label 標籤的文字顯示就是一個典型的 UI 觀察者。它在響應事件時，只會處理 next 事件，並且更新 UI 的操做須要在主線程上執行。那麼這種狀況下更好的方案就是使用 Binder改用 Binder 會簡單許多。

使用Binder改造後的代碼以下：

import UIKit
import RxSwift
import RxCocoa
 
class ViewController: UIViewController {
     
    @IBOutlet weak var label: UILabel!
    let disposeBag = DisposeBag()
  
    override func viewDidLoad() {      
        //觀察者
        let observer: Binder<String> = Binder(label) { (view, text) in
            //收到發出的索引數後顯示到label上
            view.text = text
        }
        //Observable序列（每隔1秒鐘發出一個索引數）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        observable
            .map { "當前索引數：\($0 )"}
            .bind(to: observer)
            .disposed(by: disposeBag)
    }
}
複製代碼

使用工廠方法建立Observable

在Observable的擴展類裏提供了一系列建立被觀察對象的工廠化方法。主要有下面這些：

never() ：建立一個Never序列，該序列不會發出任何事件，也不會終止。
empty(): 建立一個Empty序列，該序列只發出completed事件。
error(): 建立一個錯誤的序列，該序列以’error’事件終止。即建立一個不會發送任何條目而且當即終止錯誤的Observerable序列
just(): 建立一個Just序列，該序列只包含一個元素。
of(): 建立一個新的被觀察序列的對象，它包含可變數量的元素。
from(): 經過數組來建立一個被觀察序列。從一個序列（如Array/Dictionary/Set）中建立一個Observer
range():在指定範圍內生成一個被觀察的整數序列，發出事件n次。即建立一個Observable序列，它會發出一系列連續的整數，而後終止。
repeatElement(): 生成一個被觀察的序列，重複發出指定元素n次。
generate(): 建立一個被觀察的序列，只要提供的條件爲真，就發出狀態值。
deferred(): 爲每一個訂閱事件的觀察者都建立一個新的被觀察的序列。（一對一的關係）
doOn(): 在訂閱的被觀察者的事件執行以前，先執行do後面和要執行的訂閱事件對應的方法。
create(): 經過指定的方法實現來自定義一個被觀察的序列。
timer(): 獲取計時器Observable序列
interval(): 底層就是封裝timer

針對上面的工廠方法，下面來舉例說明

never建立序列

never()方法建立一個永遠不會發出 Event（也不會終止）的 Observable 序列。 實例4：代碼：

let disposeBag = DisposeBag()
let neverSequence = Observable<String>.never()
_ = neverSequence.subscribe {_ in
    print("This will never be printed")
}.addDisposableTo(disposeBag)
複製代碼

結果：

不會輸出任何結果。
複製代碼

empty建立序列

empty()方法建立一個空內容的 Observable 序列。 實例5：代碼：

let disposeBag = DisposeBag()
    Observable<Int>.empty().subscribe {
        event in
        print(event)
    }.addDisposableTo(disposeBag)
複製代碼

結果：

completed
複製代碼

error建立序列

error()方法建立一個不作任何操做，而是直接發送一個錯誤的Observable 序列。 實例6：代碼：

let disposeBag = DisposeBag()
    Observable<Int>.error(TestError.test)
        .subscribe { print($0) }
        .addDisposableTo(disposeBag)
複製代碼

結果：

error(test)
複製代碼

just建立序列

just()方法經過傳入一個默認值來初始化。

實例7：代碼：

let disposeBag = DisposeBag()
Observable.just("1").subscribe { event in
            print(event)
        }.addDisposableTo(disposeBag)
複製代碼

結果：

next(1)
completed
複製代碼

of建立序列

of()方法能夠接受可變數量的參數（必須要是同類型的） 實例8：代碼：

let disposeBag = DisposeBag()
Observable.of("1", "2", "3", "4").subscribe(onNext: { element in
     print(element)
}).addDisposableTo(disposeBag)
複製代碼

結果：

1
2
3
4
複製代碼

from建立序列

from()方法須要一個數組參數。

實例9：代碼：

let disposeBag = DisposeBag()
Observable.from(["1", "2", "3", "4"]).subscribe(onNext: { print($0) }).addDisposableTo(disposeBag)
複製代碼

結果：

1
2
3
4
複製代碼

range建立序列

range()方法經過指定起始和結束數值，建立一個以這個範圍內全部值做爲初始值的Observable序列。 實例10：代碼：

let disposeBag = DisposeBag()
Observable.range(start: 1, count: 10).subscribe { print($0) }.addDisposableTo(disposeBag)
複製代碼

結果：

next(1)
next(2)
next(3)
next(4)
next(5)
next(6)
next(7)
next(8)
next(9)
next(10)
completed
複製代碼

repeatElement建立序列

repeatElement()方法建立一個能夠無限發出給定元素的 Event的 Observable 序列（永不終止）。 實例11：代碼：

let disposeBag = DisposeBag()
Observable.repeatElement("1")
.take(3)
.subscribe(onNext: { print($0) })
.addDisposableTo(disposeBag)
複製代碼

結果：

1
1
1
複製代碼

generate建立序列

generate()方法建立一個只有當提供的全部的判斷條件都爲 true 的時候，纔會給出動做的 Observable 序列。 實例12：代碼：

let disposeBag = DisposeBag()
Observable.generate(
           initialState: 0,
           condition: { $0 < 3 },
           iterate: { $0 + 1 }
        )
        .subscribe(onNext: { print($0) })
        .addDisposableTo(disposeBag)
複製代碼

結果：

0
1
2
複製代碼

deferred建立序列

實例13：代碼：

let disposeBag = DisposeBag()
    var count = 1
    let deferredSequence = Observable<String>.deferred {
        print("Creating \(count)")
        count += 1

        return Observable.create { observer in
            print("Emitting...")
            observer.onNext("1")
            observer.onNext("2")
            observer.onNext("3")
            return Disposables.create()
        }
    }

    deferredSequence
        .subscribe(onNext: { print($0) })
        .addDisposableTo(disposeBag)

    deferredSequence
        .subscribe(onNext: { print($0) })
        .addDisposableTo(disposeBag)
複製代碼

結果：

Creating 1
Emitting...
1
2
3
Creating 2
Emitting...
1
2
3
複製代碼

deferred建立序列

deferred()方法至關因而建立一個 Observable 工廠，經過傳入一個 block 來執行延遲 Observable序列建立的行爲，而這個 block 裏就是真正的實例化序列對象的地方。

實例14：代碼：

let disposeBag = DisposeBag()
    var count = 1
    let deferredSequence = Observable<String>.deferred {
        print("Creating \(count)")
        count += 1

        return Observable.create { observer in
            print("Emitting...")
            observer.onNext("1")
            observer.onNext("2")
            observer.onNext("3")
            return Disposables.create()
        }
    }

    deferredSequence
        .subscribe(onNext: { print($0) })
        .addDisposableTo(disposeBag)

    deferredSequence
        .subscribe(onNext: { print($0) })
        .addDisposableTo(disposeBag)
複製代碼

結果：

Creating 1
Emitting...
1
2
3
Creating 2
Emitting...
1
2
3
複製代碼

doOn建立序列

實例15：代碼：

let disposeBag = DisposeBag()
Observable.of("1", "2", "3", "4")
          .do(onNext: { print("Intercepted:", $0) }, 
              onError { print("Intercepted error:", $0) },
              onCompleted: { print("Completed")  })
          .subscribe(onNext { print($0) },
                     onCompleted: { print("結束") })
          .addDisposableTo(disposeBag)
複製代碼

結果：

Intercepted: 1
1
Intercepted: 2
2
Intercepted: 3
3
Intercepted: 4
4
Completed
結束
複製代碼

create建立序列

create()方法接受一個 block 形式的參數，任務是對每個過來的訂閱進行處理。代碼： 實例16：

let disposeBag = DisposeBag()
let myJust = { (element: String) -> Observable<String> in
    return Observable.create { observer in  
    observer.on(.next(element))
    observer.on(.completed)
    return Disposables.create()
    }
}
myJust("1").subscribe { print($0) }.addDisposableTo(disposeBag)
複製代碼

結果：

next(1)
completed
複製代碼

timer建立序列

timer()方法有兩種用法，一種是建立的 Observable序列在通過設定的一段時間後，產生惟一的一個元素。另外一種是建立的 Observable 序列在通過設定的一段時間後，每隔一段時間產生一個元素。

實例17：

方式一：
//5秒種後發出惟一的一個元素0
let observable = Observable<Int>.timer(5, scheduler: MainScheduler.instance)
observable.subscribe { event in
    print(event)
}

方式二：
//延時5秒種後，每隔1秒鐘發出一個元素
let observable = Observable<Int>.timer(5, period: 1, scheduler: MainScheduler.instance)
observable.subscribe { event in
    print(event)
}
複製代碼

interval建立序列

interval()方法建立的 Observable 序列每隔一段設定的時間，會發出一個索引數的元素。並且它會一直髮送下去。

實例18：

let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
observable.subscribe { event in
    print(event)
}
複製代碼

Observer訂閱

有了 Observable，咱們還要使用 subscribe() 方法來訂閱它，接收它發出的 Event。

訂閱Observable有兩種方式：

咱們使用 subscribe() 訂閱了一個Observable 對象，該方法的 block 的回調參數就是被髮出的 event 事件，若是想要獲取到這個事件裏的數據，能夠經過 event.element 獲得。例如：

let observable = Observable.of("A", "B", "C")
observable.subscribe { event in
    print(event)
    //print(event.element)
}
複製代碼

結果： next(A) next(B) next(C) completed

RxSwift 還提供了另外一個 subscribe方法，它能夠把 event 進行分類,經過不一樣的 block 回調處理不一樣類型的 event.同時會把 event 攜帶的數據直接解包出來做爲參數，方便咱們使用。subscribe() 方法的 onNext、onError、onCompleted 和 onDisposed 這四個回調 block 參數都是有默認值的，即它們都是可選的。因此咱們也能夠只處理 onNext而無論其餘的狀況。

實例19 代碼以下：

let observable = Observable.of("A", "B", "C")
         
observable.subscribe(onNext: { element in
    print(element)
}, onError: { error in
    print(error)
}, onCompleted: {
    print("completed")
}, onDisposed: {
    print("disposed")
})
複製代碼

Observable銷燬

一個 Observable 序列被建立出來後它不會立刻就開始被激活從而發出 Event，而是要等到它被某我的訂閱了纔會激活它。而Observable 序列激活以後要一直等到它發出了.error或者 .completed的 event 後，它才被終結。dispose()就是用來銷燬Observable。

dispose() （1）使用該方法咱們能夠手動取消一個訂閱行爲。（2）若是咱們以爲這個訂閱結束了再也不須要了，就能夠調用 dispose()方法把這個訂閱給銷燬掉，防止內存泄漏。（3）當一個訂閱行爲被dispose 了，那麼以後 observable 若是再發出 event，這個已經 dispose 的訂閱就收不到消息了。

實例20 代碼：

let observable = Observable.of("A", "B", "C")
         
//使用subscription常量存儲這個訂閱方法
let subscription = observable.subscribe { event in
    print(event)
}
         
//調用這個訂閱的dispose()方法
subscription.dispose()
複製代碼

DisposeBag 除了 dispose()方法以外，咱們更常常用到的是一個叫 DisposeBag 的對象來管理多個訂閱行爲的銷燬：

咱們能夠把一個 DisposeBag對象當作一個垃圾袋，把用過的訂閱行爲都放進去。
而這個DisposeBag 就會在本身快要dealloc 的時候，對它裏面的全部訂閱行爲都調用 dispose()方法。

實例21：

代碼：

let disposeBag = DisposeBag()
         
//第1個Observable，及其訂閱
let observable1 = Observable.of("A", "B", "C")
observable1.subscribe { event in
    print(event)
}.disposed(by: disposeBag)
 
//第2個Observable，及其訂閱
let observable2 = Observable.of(1, 2, 3)
observable2.subscribe { event in
    print(event)
}.disposed(by: disposeBag)
複製代碼

RxSwift （三）Observable的建立，訂閱，銷燬

可觀察的序列Observable

Observable定義

Observable的建立，訂閱，銷燬

Observable建立

在 subscribe 方法中建立Observable

在 bind 方法中建立Observable

使用 AnyObserver 建立Observable

使用 Binder 建立Observable

使用工廠方法建立Observable

never建立序列

empty建立序列

error建立序列

just建立序列

of建立序列

from建立序列

range建立序列

repeatElement建立序列

generate建立序列

deferred建立序列

deferred建立序列

doOn建立序列

create建立序列

timer建立序列

interval建立序列

Observer訂閱

Observable銷燬