RxSwift入坑解讀-你所需要知道的各種概念

njhu5107 8年前發布 | 22K 次閱讀 RxSwift PEAR

相信大家很早就聽說過函數式響應編程概念，我是去年面試的時候接觸到函數式響應編程的，當時也是第一次接觸到MVVM這個概念，轉眼都一年過去了，我卻沒有在函數式編程上做深入的研究，說來還真是慚愧。

不過最近由于想要使用RxSwift，所以趁這個時候好好接觸和研究一下傳說中的函數式編程，由于網上關于RxSwift的教程資料很少，這篇文章其實就是最RxSwift官方文檔和一些概念做一些解讀。算是大家學習的參考文章吧！先挖個坑，這可能會是一個RxSwift系列，希望大家在學習的時候有所參考。

RxSwift是什么

RxSwif是 ReactiveX 的Swift版本，也就是一個函數式響應編程的框架。對，就這一句話。想要知道他做什么的，我們先來了解一下觀察者模式。

觀察者模式

關于觀察者模式我想大伙應該都很了解了吧，什么KVO，通知等都是觀察者模式，在設計模式中他可是一個重中之重的設計模式啊！比如一個寶寶在睡覺，爸爸媽媽，爺爺奶奶總不能在那邊一只看著吧？那樣子太累了。他們該做啥事就做啥事唄，只要聽到寶寶的哭聲，他們就給寶寶喂奶就行了。這就是一個典型的觀察者模式。寶寶是被觀察者，爸爸媽媽等是觀察者也稱作訂閱者，只要被觀察者發出了某些事件比如寶寶哭聲、叫聲都是一個事件，通知到訂閱者，訂閱者們就可以做相應的處理工作。哈哈，觀察者模式很簡單吧？

RxSwift做了什么

RxSwift把我們程序中每一個操作都看成一個事件，比如一個TextField中的文本改變，一個按鈕被點擊，或者一個網絡請求結束等，每一個事件源就可以看成一個管道，也就是sequence，比如TextField，當我們改變里面的文本的時候，這個TextField就會不斷的發出事件，從他的這個sequence中不斷的流出，我們只需要監聽這個sequence，每流出一個事件就做相應的處理。同理，Button也是一個sequence，每點擊一次就流出一個事件。也就是我們把每一步都想成是一個事件就好去理解RxSwift了。看下圖是不是很好理解了？

sequence.png

Observable和Observer

理解了觀察者模式這兩個概念就很好理解了，Observable就是可被觀察的，也就是我們說的寶寶，他也是事件源。而Observer就是我們的觀察者，也就是當收到事件的時候去做某些處理的爸爸媽媽。觀察者需要去訂閱(subscribe)被觀察者，才能收到Observable的事件通知消息。

下面開始一些基本概念解讀，通讀一遍你會對RxSwift有非常深刻的認識了，其實也就是對整理了一下官方文檔和加上自己的一些理解

創建和訂閱被觀察者

下面創建被觀察者其實就是創建一個Obserable的sequence，就是創建一個流，然后就可以被訂閱subscribe，這樣被觀察者發出時間消失，我們就能做相應的處理

DisposeBag

DisposeBag其實就相當于iOS中的ARC似得，會在適當的時候銷毀觀察者，相當于內存管理者吧。

subscribe是訂閱sequence發出的事件，比如next事件，error事件等。而subscribe(onNext:)是監聽sequence發出的next事件中的element進行處理，他會忽略error和completed事件。相對應的還有subscribe(onError:) 和 subscribe(onCompleted:)

never

never就是創建一個sequence，但是不發出任何事件信號。

never.png

let disposeBag = DisposeBag()
let neverSequence = Observable<String>.never()

let neverSequenceSubscription = neverSequence
    .subscribe { _ in
        print("This will never be printed")
}.addDisposableTo(disposeBag)

結果是什么都不打印

empty

empty就是創建一個空的sequence,只能發出一個completed事件

empty.png

let disposeBag = DisposeBag()

    Observable<Int>.empty()
        .subscribe { event in
            print(event)
        }
        .addDisposableTo(disposeBag)

completed

just

just是創建一個sequence只能發出一種特定的事件，能正常結束

just.png

let disposeBag = DisposeBag()

Observable.just(":red_circle:")
    .subscribe { event in
        print(event)
    }
    .addDisposableTo(disposeBag)

next(:red_circle:)
completed

of

of是創建一個sequence能發出很多種事件信號

let disposeBag = DisposeBag()

Observable.of(":dog:", ":cat:", ":mouse:", ":hamster:")
    .subscribe(onNext: { element in
        print(element)
    })
    .addDisposableTo(disposeBag)

:dog:
:cat:
:mouse:
:hamster:

如果把上面的onNext:去掉的話，結果會是這樣子，也正好對應了我們subscribe中，subscribe只監聽事件。

next(:dog:)
next(:cat:)
next(:mouse:)
next(:hamster:)
completed

from

from就是從集合中創建sequence，例如數組，字典或者Set

let disposeBag = DisposeBag()

Observable.from([":dog:", ":cat:", ":mouse:", ":hamster:"])
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

create

我們也可以自定義可觀察的sequence，那就是使用create

create.png

create操作符傳入一個觀察者observer，然后調用observer的onNext，onCompleted和onError方法。返回一個可觀察的obserable序列。

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(":red_circle:")
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

next(:red_circle:)
completed

range

range就是創建一個sequence，他會發出這個范圍中的從開始到結束的所有事件

range.png

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

創建一個sequence，發出特定的事件n次

repeat.png

let disposeBag = DisposeBag()

Observable.repeatElement(":red_circle:")
    .take(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:red_circle:
:red_circle:
:red_circle:

generate

generate是創建一個可觀察sequence，當初始化的條件為true的時候，他就會發出所對應的事件

let disposeBag = DisposeBag()

Observable.generate(
        initialState: 0,
        condition: { $0 < 3 },
        iterate: { $0 + 1 }
    )
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

deferred

deferred會為每一為訂閱者observer創建一個新的可觀察序列

defered.png

下面例子中每次進行subscribe的時候都會去創建一個新的deferredSequence，所以Emitting會打印兩遍。

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(":dog:")
        observer.onNext(":cat:")
        observer.onNext(":monkey_face:")
        return Disposables.create()
    }
}

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

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

Creating 1
Emitting...
:dog:
:cat:
:monkey_face:
Creating 2
Emitting...
:dog:
:cat:
:monkey_face:

error

創建一個可觀察序列，但不發出任何正常的事件，只發出error事件并結束

let disposeBag = DisposeBag()

Observable<Int>.error(TestError.test)
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

error(test)

doOn

doOn我感覺就是在直接onNext處理時候，先執行某個方法，doOnNext( :)方法就是在subscribe(onNext:)前調用，doOnCompleted( :)就是在subscribe(onCompleted:)前面調用的。

let disposeBag = DisposeBag()

Observable.of(":apple:", ":pear:", ":tangerine:", ":lemon:")
    .do(onNext: { print("Intercepted:", $0) }, onError: { print("Intercepted error:", $0) }, onCompleted: { print("Completed")  })
    .subscribe(onNext: { print($0) },onCompleted: { print("結束") })
    .addDisposableTo(disposeBag)

Intercepted: :apple:
:apple:
Intercepted: :pear:
:pear:
Intercepted: :tangerine:
:tangerine:
Intercepted: :lemon:
:lemon:
Completed
結束

學會使用Subjects

Subjet是observable和Observer之間的橋梁，一個Subject既是一個Obserable也是一個Observer，他既可以發出事件，也可以監聽事件。

PublishSubject

當你訂閱PublishSubject的時候，你只能接收到訂閱他之后發生的事件。subject.onNext()發出onNext事件，對應的還有onError()和onCompleted()事件

publishsubject.png

let disposeBag = DisposeBag()
let subject = PublishSubject<String>()

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext(":dog:")
subject.onNext(":cat:")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext(":a:")
subject.onNext(":b:")

Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)

ReplaySubject

當你訂閱ReplaySubject的時候，你可以接收到訂閱他之后的事件，但也可以接受訂閱他之前發出的事件，接受幾個事件取決與bufferSize的大小

replaysubject.png

let disposeBag = DisposeBag()
let subject = ReplaySubject<String>.create(bufferSize: 1)

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext(":dog:")
subject.onNext(":cat:")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext(":a:")
subject.onNext(":b:")

Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 2 Event: next(:cat:) //訂閱之后還可以接受一次前面發出的事件
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)

BehaviorSubject

當你訂閱了BehaviorSubject，你會接受到訂閱之前的最后一個事件。

behaviorsubject.png

let disposeBag = DisposeBag()
let subject = BehaviorSubject(value: ":red_circle:")

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext(":dog:")
subject.onNext(":cat:")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext(":a:")
subject.onNext(":b:")

subject.addObserver("3").addDisposableTo(disposeBag)
subject.onNext(":pear:")
subject.onNext(":tangerine:")

Subscription: 1 Event: next(:red_circle:)
Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 2 Event: next(:cat:)    //訂閱之前的最后一個事件
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)
Subscription: 3 Event: next(:b:) //訂閱之前的最后一個事件
Subscription: 1 Event: next(:pear:)
Subscription: 3 Event: next(:pear:)
Subscription: 2 Event: next(:pear:)
Subscription: 1 Event: next(:tangerine:)
Subscription: 3 Event: next(:tangerine:)
Subscription: 2 Event: next(:tangerine:)

PublishSubject, ReplaySubject和BehaviorSubject是不會自動發出completed事件的。

Variable

Variable是BehaviorSubject一個包裝箱，就像是一個箱子一樣，使用的時候需要調用asObservable()拆箱，里面的value是一個BehaviorSubject，他不會發出error事件，但是會自動發出completed事件。

let disposeBag = DisposeBag()
let variable = Variable(":red_circle:")

variable.asObservable().addObserver("1").addDisposableTo(disposeBag)
variable.value = ":dog:"
variable.value = ":cat:"

variable.asObservable().addObserver("2").addDisposableTo(disposeBag)
variable.value = ":a:"
variable.value = ":b:"

Subscription: 1 Event: next(:red_circle:)
Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 2 Event: next(:cat:)
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)
Subscription: 1 Event: completed
Subscription: 2 Event: completed

聯合操作

聯合操作就是把多個Observable流合成單個Observable流

startWith

在發出事件消息之前，先發出某個特定的事件消息。比如發出事件2 ，3然后我startWith(1)，那么就會先發出1，然后2 ，3.

startwith.png

let disposeBag = DisposeBag()

Observable.of("2", "3")
    .startWith("1")
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

merge

合并兩個Observable流合成單個Observable流，根據時間軸發出對應的事件

merge.png

let disposeBag = DisposeBag()

let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()

Observable.of(subject1, subject2)
    .merge()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

subject1.onNext(":a:")

subject1.onNext(":b:")

subject2.onNext("①")

subject2.onNext("②")

subject1.onNext(":ab:")

subject2.onNext("③")

:a:
:b:
①
②
:ab:
③

zip

綁定超過最多不超過8個的Observable流，結合在一起處理。注意Zip是一個事件對應另一個流一個事件。

zip.png

let disposeBag = DisposeBag()

let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()

Observable.zip(stringSubject, intSubject) { stringElement, intElement in
    "\(stringElement) \(intElement)"
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

stringSubject.onNext(":a:")
stringSubject.onNext(":b:")

intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext(":ab:")
intSubject.onNext(3)

:a: 1    將stringSubject和intSubject壓縮到一起共同處理
:b: 2
:ab: 3

combineLatest

綁定超過最多不超過8個的Observable流，結合在一起處理。和Zip不同的是combineLatest是一個流的事件對應另一個流的最新的事件，兩個事件都會是最新的事件，可將下圖與Zip的圖進行對比。

combinlatest.png

let disposeBag = DisposeBag()

let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()

Observable.combineLatest(stringSubject, intSubject) { stringElement, intElement in
        "\(stringElement) \(intElement)"
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

stringSubject.onNext(":a:")

stringSubject.onNext(":b:")
intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext(":ab:")

:b: 1
:b: 2
:ab: 2

switchLatest

switchLatest可以對事件流進行轉換，本來監聽的subject1，我可以通過更改variable里面的value更換事件源。變成監聽subject2了

switch.png

let disposeBag = DisposeBag()

let subject1 = BehaviorSubject(value: ":soccer:?")
let subject2 = BehaviorSubject(value: ":apple:")

let variable = Variable(subject1)

variable.asObservable()
    .switchLatest()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

subject1.onNext(":football:")
subject1.onNext(":basketball:")

variable.value = subject2

subject1.onNext(":baseball:")

subject2.onNext(":pear:")
variable.value = subject1
subject2.onNext("田騰飛")
subject1.onNext("沸騰天")

:soccer:?
:football:
:basketball:
:apple:
:pear:
:baseball:
沸騰天

變換操作

map

通過傳入一個函數閉包把原來的sequence轉變為一個新的sequence的操作

transform.png

let disposeBag = DisposeBag()
Observable.of(1, 2, 3)
    .map { $0 * $0 }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

1    每一個元素自己相乘
4
9

flatMap

將一個sequence轉換為一個sequences，當你接收一個sequence的事件，你還想接收其他sequence發出的事件的話可以使用flatMap，她會將每一個sequence事件進行處理以后，然后再以一個sequence形式發出事件。而且flatMap有一次拆包動作，請看代碼解析。

flatmap.png

let disposeBag = DisposeBag()

struct Player {
    var score: Variable<Int>        //里面是一個Variable
}

let :boy|type_1_2: = Player(score: Variable(80))        
let :girl|type_3: = Player(score: Variable(90))
let :joy: = Player(score: Variable(550))

let player = Variable(:boy|type_1_2:)  //將player轉為Variable

player.asObservable()        //拆箱轉成可被監聽的sequence
    .flatMap { $0.score.asObservable() } // flatMap有一次拆包動作，$0本來應該是一個BehaviorSubject類型，但是直接訪問了score。所以猜想flatMap對behaviorSubject進行了onNext拆包取數據
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:boy|type_1_2:.score.value = 85

player.value = :girl|type_3: //更換了value，相當于又添加了一個sequence，兩個sequence都可以接收

:boy|type_1_2:.score.value = 95
:boy|type_1_2:.score.value = 222
player.value = :joy:

:girl|type_3:.score.value = 100

flatMapLatest

flatMapLatest只會接收最新的value事件，將上例改為flatMapLatest。結果為

scan

scan就是給一個初始化的數，然后不斷的拿前一個結果和最新的值進行處理操作。

let disposeBag = DisposeBag()

Observable.of(10, 100, 1000)
    .scan(1) { aggregateValue, newValue in
        aggregateValue + newValue
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

過濾和約束

filter

filter很好理解，就是過濾掉某些不符合要求的事件

let disposeBag = DisposeBag()

Observable.of(
    ":cat:", ":rabbit:", ":dog:",
    ":frog:", ":cat:", ":rabbit:",
    ":hamster:", ":frog:", ":cat:")
    .filter {
        $0 == ":cat:"
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:cat:
:cat:
:cat:

distinctUntilChanged

distinctUntilChanged就是當下一個事件與前一個事件是不同事件的事件才進行處理操作

let disposeBag = DisposeBag()

Observable.of(":cat:", ":pig:", ":cat:", ":cat:", ":cat:", ":monkey_face:", ":cat:")
    .distinctUntilChanged()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:cat:
:pig:
:cat:
:monkey_face:
:cat:

elementAt

只處理在指定位置的事件

let disposeBag = DisposeBag()

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .elementAt(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:frog:

single

找出在sequence只發出一次的事件，如果超過一個就會發出error錯誤

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .single()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:cat:    //單一信號超過了一個
Received unhandled error: /var/folders/hz/v15ld5mj0nqf83d21j13y0tw0000gn/T/./lldb/7229/playground107.swift:69:__lldb_expr_107 -> Sequence contains more than one element.

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .single { $0 == ":frog:" }        //青蛙只有一個，completed
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

Observable.of(":cat:", ":rabbit:", ":dog:", ":cat:", ":rabbit:", ":dog:")
    .single { $0 == ":rabbit:" } //兔子有兩個，會發出error
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .single { $0 == ":large_blue_circle:" } //沒有藍色球，會發出error
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

take

只處理前幾個事件信號,

let disposeBag = DisposeBag()

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .take(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:cat:
:rabbit:
:dog:

takeLast

只處理后幾個事件信號

let disposeBag = DisposeBag()

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .takeLast(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:frog:
:pig:
:monkey_face:

takeWhile

當條件滿足的時候進行處理

let disposeBag = DisposeBag()

Observable.of(1, 2, 3, 4, 5, 6)
    .takeWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

takeUntil

接收事件消息，直到另一個sequence發出事件消息的時候。

let disposeBag = DisposeBag()

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()

sourceSequence
    .takeUntil(referenceSequence)
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

sourceSequence.onNext(":cat:")
sourceSequence.onNext(":rabbit:")
sourceSequence.onNext(":dog:")

referenceSequence.onNext(":red_circle:")    //停止接收消息

sourceSequence.onNext(":frog:")
sourceSequence.onNext(":pig:")
sourceSequence.onNext(":monkey_face:")

next(:cat:)
next(:rabbit:)
next(:dog:)
completed

skip

取消前幾個事件

let disposeBag = DisposeBag()

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .skip(2)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

:dog:
:frog:
:pig:
:monkey_face:

skipWhile

滿足條件的事件消息都取消

let disposeBag = DisposeBag()

Observable.of(1, 2, 3, 4, 5, 6)
    .skipWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

skipWhileWithIndex

滿足條件的都被取消，傳入的閉包同skipWhile有點區別而已

let disposeBag = DisposeBag()

Observable.of(":cat:", ":rabbit:", ":dog:", ":frog:", ":pig:", ":monkey_face:")
    .skipWhileWithIndex { element, index in
        index < 3
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

skipUntil

直到某個sequence發出了事件消息，才開始接收當前sequence發出的事件消息

let disposeBag = DisposeBag()

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()

sourceSequence
    .skipUntil(referenceSequence)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

sourceSequence.onNext(":cat:")
sourceSequence.onNext(":rabbit:")
sourceSequence.onNext(":dog:")

referenceSequence.onNext(":red_circle:")

sourceSequence.onNext(":frog:")
sourceSequence.onNext(":pig:")
sourceSequence.onNext(":monkey_face:")
}

數學操作

toArray

將sequence轉換成一個array，并轉換成單一事件信號，然后結束

let disposeBag = DisposeBag()

Observable.range(start: 1, count: 10)
    .toArray()
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

next([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
completed

reduce

用一個初始值，對事件數據進行累計操作。reduce接受一個初始值，和一個操作符號

let disposeBag = DisposeBag()

Observable.of(10, 100, 1000)
    .reduce(1, accumulator: +)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

concat

concat會把多個sequence和并為一個sequence，并且當前面一個sequence發出了completed事件，才會開始下一個sequence的事件。

在第一sequence完成之前，第二個sequence發出的事件都會被忽略，但會接收一完成之前的二發出的最后一個事件。不好解釋，看例子說明

let disposeBag = DisposeBag()

let subject1 = BehaviorSubject(value: ":apple:")
let subject2 = BehaviorSubject(value: ":dog:")

let variable = Variable(subject1)

variable.asObservable()
    .concat()
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

subject1.onNext(":pear:")
subject1.onNext(":tangerine:")

variable.value = subject2


subject2.onNext(":cat:")    //1完成前，會被忽略
subject2.onNext("teng") //1完成前，會被忽略
subject2.onNext("fei")    //1完成前的最后一個，會被接收

subject1.onCompleted()

subject2.onNext(":mouse:")

next(:apple:)
next(:pear:)
next(:tangerine:)
next(fei)
next(:mouse:)

連接性操作

Connectable Observable有訂閱時不開始發射事件消息，而是僅當調用它們的connect（）方法時。這樣就可以等待所有我們想要的訂閱者都已經訂閱了以后，再開始發出事件消息，這樣能保證我們想要的所有訂閱者都能接收到事件消息。其實也就是等大家都就位以后，開始發出消息。

publish

將一個正常的sequence轉換成一個connectable sequence

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

_ = intSequence
    .subscribe(onNext: { print("Subscription 1:, Event: \($0)") })

delay(2) { _ = intSequence.connect() } //相當于把事件消息推遲了兩秒

delay(4) {
    _ = intSequence
        .subscribe(onNext: { print("Subscription 2:, Event: \($0)") })
}

delay(6) {
    _ = intSequence
        .subscribe(onNext: { print("Subscription 3:, Event: \($0)") })
}

Subscription 1:, Event: 0
Subscription 1:, Event: 1
Subscription 2:, Event: 1
Subscription 1:, Event: 2
Subscription 2:, Event: 2
Subscription 1:, Event: 3
Subscription 3:, Event: 3
Subscription 2:, Event: 3
Subscription 1:, Event: 4
Subscription 3:, Event: 4

replay

將一個正常的sequence轉換成一個connectable sequence，然后和replaySubject相似，能接收到訂閱之前的事件消息。

let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
    .replay(5)    //接收到訂閱之前的5條事件消息

_ = intSequence
    .subscribe(onNext: { print("Subscription 1:, Event: \($0)") })

delay(2) { _ = intSequence.connect() }

delay(4) {
    _ = intSequence
        .subscribe(onNext: { print("Subscription 2:, Event: \($0)") })
}

delay(8) {
    _ = intSequence
        .subscribe(onNext: { print("Subscription 3:, Event: \($0)") })
}

multicast

將一個正常的sequence轉換成一個connectable sequence，并且通過特性的subject發送出去，比如PublishSubject，或者replaySubject，behaviorSubject等。不同的Subject會有不同的結果。

let subject = PublishSubject<Int>()

_ = subject
    .subscribe(onNext: { print("Subject: \($0)") })

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

_ = intSequence
    .subscribe(onNext: { print("\tSubscription 1:, Event: \($0)") })

delay(2) { _ = intSequence.connect() }

delay(4) {
    _ = intSequence
        .subscribe(onNext: { print("\tSubscription 2:, Event: \($0)") })
}

錯誤處理

catchErrorJustReturn

遇到error事件的時候，就return一個值，然后結束

let disposeBag = DisposeBag()

let sequenceThatFails = PublishSubject<String>()

sequenceThatFails
    .catchErrorJustReturn(":blush:")
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

sequenceThatFails.onNext(":grimacing:")
sequenceThatFails.onNext(":fearful:")
sequenceThatFails.onNext(":rage:")
sequenceThatFails.onNext(":red_circle:")
sequenceThatFails.onError(TestError.test)

next(:grimacing:)
next(:fearful:)
next(:rage:)
next(:red_circle:)
next(:blush:)
completed

catchError

捕獲error進行處理，可以返回另一個sequence進行訂閱

let disposeBag = DisposeBag()

let sequenceThatFails = PublishSubject<String>()
let recoverySequence = PublishSubject<String>()

sequenceThatFails
    .catchError {
        print("Error:", $0)
        return recoverySequence
    }
    .subscribe { print($0) }
    .addDisposableTo(disposeBag)

sequenceThatFails.onNext(":grimacing:")
sequenceThatFails.onNext(":fearful:")
sequenceThatFails.onNext(":rage:")
sequenceThatFails.onNext(":red_circle:")
sequenceThatFails.onError(TestError.test)

recoverySequence.onNext(":blush:")

next(:grimacing:)
next(:fearful:)
next(:rage:)
next(:red_circle:)
Error: test
next(:blush:)

retry

遇見error事件可以進行重試，比如網絡請求失敗，可以進行重新連接

let disposeBag = DisposeBag()
var count = 1

let sequenceThatErrors = Observable<String>.create { observer in
    observer.onNext(":apple:")
    observer.onNext(":pear:")
    observer.onNext(":tangerine:")

    if count == 1 {
        observer.onError(TestError.test)
        print("Error encountered")
        count += 1
    }

    observer.onNext(":dog:")
    observer.onNext(":cat:")
    observer.onNext(":mouse:")
    observer.onCompleted()

    return Disposables.create()
}

sequenceThatErrors
    .retry(3)        //不傳入數字的話，只會重試一次
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

debug

打印所有的訂閱, 事件和disposals

sequenceThatErrors
    .retry(3)
    .debug()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

RxSwift.Resources.total

查看RxSwift所有資源的占用

print(RxSwift.Resources.total)

啊，文章終于結束，這篇文章比較長，基本上涵蓋了官方文檔所有的概念，其中不免有些錯誤與疏漏，希望能在你學習RxSwift的時候能有一些參考價值吧！！！

來自：http://www.jianshu.com/p/a1e2665f9a6c

本文由用戶 njhu5107 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1481854306379.html

RxSwift PEAR

RxSwift入坑解讀-你所需要知道的各種概念

RxSwift是什么

觀察者模式

RxSwift做了什么

Observable和Observer

創建和訂閱被觀察者

DisposeBag

subscribe

never

empty

just

of

from

create

range

repeatElement

generate

deferred

error

doOn

學會使用Subjects

PublishSubject

ReplaySubject

BehaviorSubject

Variable

聯合操作

startWith

merge

zip

combineLatest

switchLatest

變換操作

map

flatMap

flatMapLatest

scan

過濾和約束

filter

distinctUntilChanged

elementAt

single

take

takeLast

takeWhile

takeUntil

skip

skipWhile

skipWhileWithIndex

skipUntil

數學操作

toArray

reduce

concat

連接性操作

publish

replay

multicast

錯誤處理

catchErrorJustReturn

catchError

retry

debug

debug

RxSwift.Resources.total

相關經驗

相關資訊

相關文檔

目錄