RxSwift入坑記-各種概念解讀

ShaFlinchum 9年前發布 | 14K 次閱讀 RxSwift 設計模式

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

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

RxSwift是什么

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

觀察者模式

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

RxSwift做了什么

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

Observable和Observer

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

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

創建和訂閱被觀察者

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

DisposeBag

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

subscribe

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

never

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

letdisposeBag =DisposeBag()
letneverSequence =Observable<String>.never()

letneverSequenceSubscription = neverSequence
 .subscribe { _in
print("This will never be printed")
}.addDisposableTo(disposeBag)
結果是什么都不打印

empty

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

letdisposeBag =DisposeBag()

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

just

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

letdisposeBag =DisposeBag()

Observable.just("?")
 .subscribe { event in
print(event)
 }
 .addDisposableTo(disposeBag)
next(?)
completed

of

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

letdisposeBag =DisposeBag()

Observable.of("?","?","?","?")
 .subscribe(onNext: { element in
print(element)
 })
 .addDisposableTo(disposeBag)
?
?
?
?

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

next(?)
next(?)
next(?)
next(?)
completed

from

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

letdisposeBag =DisposeBag()

Observable.from(["?","?","?","?"])
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)

create

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

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

letdisposeBag =DisposeBag()

letmyJust = { (element:String) ->Observable<String>in
returnObservable.create { observerin
 observer.on(.next(element))
 observer.on(.completed)
returnDisposables.create()
 }
}

myJust("?")
 .subscribe { print($0) }
 .addDisposableTo(disposeBag)
next(?)
completed

range

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

letdisposeBag =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次

letdisposeBag =DisposeBag()

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

generate

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

letdisposeBag =DisposeBag()

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

deferred

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

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

letdisposeBag =DisposeBag()
varcount=1

letdeferredSequence =Observable<String>.deferred {
print("Creating\(count)")
count+=1

returnObservable.create { observerin
print("Emitting...")
 observer.onNext("?")
 observer.onNext("?")
 observer.onNext("?")
returnDisposables.create()
 }
}

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

deferredSequence
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
Creating 1
Emitting...
?
?
?
Creating 2
Emitting...
?
?
?

error

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

letdisposeBag =DisposeBag()

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

doOn

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

letdisposeBag =DisposeBag()

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

學會使用Subjects

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

PublishSubject

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

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

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext("?")
subject.onNext("?")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext("??")
subject.onNext("??")
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)

ReplaySubject

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

letdisposeBag =DisposeBag()
letsubject =ReplaySubject<String>.create(bufferSize:1)

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext("?")
subject.onNext("?")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext("??")
subject.onNext("??")
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 2 Event: next(?) //訂閱之后還可以接受一次前面發出的事件
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)

BehaviorSubject

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

letdisposeBag =DisposeBag()
letsubject =BehaviorSubject(value:"?")

subject.addObserver("1").addDisposableTo(disposeBag)
subject.onNext("?")
subject.onNext("?")

subject.addObserver("2").addDisposableTo(disposeBag)
subject.onNext("??")
subject.onNext("??")

subject.addObserver("3").addDisposableTo(disposeBag)
subject.onNext("?")
subject.onNext("?")
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 2 Event: next(?) //訂閱之前的最后一個事件
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 3 Event: next(??) //訂閱之前的最后一個事件
Subscription: 1 Event: next(?)
Subscription: 3 Event: next(?)
Subscription: 2 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 3 Event: next(?)
Subscription: 2 Event: next(?)

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

Variable

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

letdisposeBag =DisposeBag()
letvariable =Variable("?")

variable.asObservable().addObserver("1").addDisposableTo(disposeBag)
variable.value = "?"
variable.value = "?"

variable.asObservable().addObserver("2").addDisposableTo(disposeBag)
variable.value = "??"
variable.value = "??"
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 1 Event: next(?)
Subscription: 2 Event: next(?)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 1 Event: next(??)
Subscription: 2 Event: next(??)
Subscription: 1 Event: completed
Subscription: 2 Event: completed

聯合操作

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

startWith

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

letdisposeBag =DisposeBag()

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

merge

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

letdisposeBag =DisposeBag()

letsubject1 =PublishSubject<String>()
letsubject2 =PublishSubject<String>()

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

subject1.onNext("??")

subject1.onNext("??")

subject2.onNext("①")

subject2.onNext("②")

subject1.onNext("?")

subject2.onNext("③")
??
??
①
②
?
③

zip

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

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("??")
stringSubject.onNext("??")

intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext("?")
intSubject.onNext(3)
?? 1 將stringSubject和intSubject壓縮到一起共同處理
?? 2
? 3

combineLatest

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

letdisposeBag =DisposeBag()

letstringSubject =PublishSubject<String>()
letintSubject =PublishSubject<Int>()

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

stringSubject.onNext("??")

stringSubject.onNext("??")
intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext("?")
?? 1
?? 2
? 2

switchLatest

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

letdisposeBag =DisposeBag()

letsubject1 =BehaviorSubject(value:"??")
letsubject2 =BehaviorSubject(value:"?")

letvariable =Variable(subject1)

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

subject1.onNext("?")
subject1.onNext("?")

variable.value = subject2

subject1.onNext("??")

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

變換操作

map

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

letdisposeBag =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有一次拆包動作,請看代碼解析。

let disposeBag = DisposeBag()

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

let ?? = Player(score: Variable(80)) 
let ?? = Player(score: Variable(90))
let ? = Player(score: Variable(550))

let player = Variable(??) //將player轉為Variable

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

??.score.value = 85

player.value = ?? //更換了value,相當于又添加了一個sequence,兩個sequence都可以接收

??.score.value = 95
??.score.value = 222
player.value = ?

??.score.value = 100

flatMapLatest

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

scan

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

letdisposeBag =DisposeBag()

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

過濾和約束

filter

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

letdisposeBag =DisposeBag()

Observable.of(
"?","?","?",
"?","?","?",
"?","?","?")
 .filter{
 $0=="?"
 }
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
?
?
?

distinctUntilChanged

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

letdisposeBag =DisposeBag()

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

elementAt

只處理在指定位置的事件

letdisposeBag =DisposeBag()

Observable.of("?","?","?","?","?","?")
 .elementAt(3)
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)

?

single

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

Observable.of("?","?","?","?","?","?")
 .single()
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
? //單一信號超過了一個
Received unhandled error: /var/folders/hz/v15ld5mj0nqf83d21j13y0tw0000gn/T/./lldb/7229/playground107.swift:69:__lldb_expr_107 -> Sequence contains more than one element.
Observable.of("?","?","?","?","?","?")
 .single { $0=="?"}//青蛙只有一個,completed
 .subscribe { print($0) }
 .addDisposableTo(disposeBag)
Observable.of("?", "?", "?", "?", "?", "?")
 .single { $0 == "?" } //兔子有兩個,會發出error
 .subscribe { print($0) }
 .addDisposableTo(disposeBag)
Observable.of("?", "?", "?", "?", "?", "?")
 .single { $0 == "?" } //沒有藍色球,會發出error
 .subscribe { print($0) }
 .addDisposableTo(disposeBag)

take

只處理前幾個事件信號,

letdisposeBag =DisposeBag()

Observable.of("?","?","?","?","?","?")
 .take(3)
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
?
?
?

takeLast

只處理后幾個事件信號

letdisposeBag =DisposeBag()

Observable.of("?","?","?","?","?","?")
 .takeLast(3)
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
?
?
?

takeWhile

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

letdisposeBag =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("?")
sourceSequence.onNext("?")
sourceSequence.onNext("?")

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

sourceSequence.onNext("?")
sourceSequence.onNext("?")
sourceSequence.onNext("?")
next(?)
next(?)
next(?)
completed

skip

取消前幾個事件

letdisposeBag =DisposeBag()

Observable.of("?","?","?","?","?","?")
 .skip(2)
 .subscribe(onNext: { print($0) })
 .addDisposableTo(disposeBag)
?
?
?
?

skipWhile

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

letdisposeBag =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("?", "?", "?", "?", "?", "?")
 .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("?")
sourceSequence.onNext("?")
sourceSequence.onNext("?")

referenceSequence.onNext("?")

sourceSequence.onNext("?")
sourceSequence.onNext("?")
sourceSequence.onNext("?")
}

數學操作

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: "?")
let subject2 = BehaviorSubject(value: "?")

let variable = Variable(subject1)

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

subject1.onNext("?")
subject1.onNext("?")

variable.value = subject2


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

subject1.onCompleted()

subject2.onNext("?")
next(?)
next(?)
next(?)
next(fei)
next(?)

連接性操作

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一個值,然后結束

letdisposeBag =DisposeBag()

letsequenceThatFails =PublishSubject<String>()

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

sequenceThatFails.onNext("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onError(TestError.test)
next(?)
next(?)
next(?)
next(?)
next(?)
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("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onNext("?")
sequenceThatFails.onError(TestError.test)

recoverySequence.onNext("?")
next(?)
next(?)
next(?)
next(?)
Error: test
next(?)

retry

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

let disposeBag = DisposeBag()
var count = 1

let sequenceThatErrors = Observable<String>.create { observer in
 observer.onNext("?")
 observer.onNext("?")
 observer.onNext("?")

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

 observer.onNext("?")
 observer.onNext("?")
 observer.onNext("?")
 observer.onCompleted()

 return Disposables.create()
}

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

debug

debug

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

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

RxSwift.Resources.total

查看RxSwift所有資源的占用

print(RxSwift.Resources.total)

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

 

來自:http://www.codertian.com/2016/11/27/RxSwift-ru-keng-ji-read-document/

 

 本文由用戶 ShaFlinchum 自行上傳分享,僅供網友學習交流。所有權歸原作者,若您的權利被侵害,請聯系管理員。
 轉載本站原創文章,請注明出處,并保留原始鏈接、圖片水印。
 本站是一個以用戶分享為主的開源技術平臺,歡迎各類分享!