RxSwift 使用

RxSwift把我们程序中每一个操作都看成一个事件,比如一个TextField中的文本改变,一个按钮被点击,或者一个网络请求结束等,每一个事件源就可以看成一个管道,也就是sequence,比如TextField,当我们改变里面的文本的时候,这个TextField就会不断的发出事件,从他的这个sequence中不断的流出,我们只需要监听这个sequence,每流出一个事件就做相应的处理。同理,Button也是一个sequence,每点击一次就流出一个事件。也就是我们把每一步都想成是一个事件就好去理解RxSwift了。看下图是不是很好理解了?

Observable和Observer

理解了观察者模式这两个概念就很好理解了,Observable就是可被观察的,也就是我们说的宝宝,他也是事件源。而Observer就是我们的观察者,也就是当收到事件的时候去做某些处理的爸爸妈妈。观察者需要去订阅(subscribe)被观察者,才能收到Observable的事件通知消息。

创建和订阅被观察者

下面创建被观察者其实就是创建一个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,但是不发出任何事件信号。

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事件

let disposeBag = DisposeBag()

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

completed

just
just是创建一个sequence只能发出一种特定的事件,能正常结束

let disposeBag = DisposeBag()

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

of是创建一个sequence能发出很多种事件信号

let disposeBag = DisposeBag()

Observable.of("🐶", "🐱", "🐭", "🐹")
    .subscribe(onNext: { element in
        print(element)
    })
    .addDisposableTo(disposeBag)
🐶
🐱
🐭

如果把上面的onNext:去掉的话,结果会是这样子,也正好对应了我们subscribe中,subscribe只监听事件。

next(🐶)
next(🐱)
next(🐭)
next(🐹)
completed

from

from就是从集合中创建sequence,例如数组,字典或者Set

let disposeBag = DisposeBag()

Observable.from(["🐶", "🐱", "🐭", "🐹"])
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

create

我们也可以自定义可观察的sequence,那就是使用create

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

range

range就是创建一个sequence,他会发出这个范围中的从开始到结束的所有事件

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

let disposeBag = DisposeBag()

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

generate

let disposeBag = DisposeBag()

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

deferred会为每一为订阅者observer创建一个新的可观察序列

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("🐶")
        observer.onNext("🐱")
        observer.onNext("🐵")
        return Disposables.create()
    }
}

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

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

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("🍎", "🍐", "🍊", "🍋")
    .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的大小

let disposeBag = DisposeBag()
let subject = 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,你会接受到订阅之前的最后一个事件。

let disposeBag = DisposeBag()
let subject = 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事件

let disposeBag = DisposeBag()
let variable = 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.

![693607-f2f8af4760c09dd122222.jpeg](http://upload-images.jianshu.io/upload_images/1935872-38d23e1661ff2d65.jpeg?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
let disposeBag = DisposeBag()

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

合并两个Observable流合成单个Observable流,根据时间轴发出对应的事件


let disposeBag = DisposeBag()

let subject1 = PublishSubject<String>()
let subject2 = 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
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的图进行对比。

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("🅰️")

stringSubject.onNext("🅱️")
intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext("🆎")

🅱️ 1
🅱️ 2
🆎 2
switchLatest

switchLatest可以对事件流进行转换,本来监听的subject1,我可以通过更改variable里面的value更换事件源。变成监听subject2了

let disposeBag = DisposeBag()

let subject1 = BehaviorSubject(value: "⚽️")
let subject2 = BehaviorSubject(value: "🍎")

let variable = 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的操作

et 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有一次拆包动作,请看代码解析。

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`
80
85
90
95
222
550
100
flatMapLatest

flatMapLatest只会接收最新的value事件,将上例改为flatMapLatest。结果为

80
85
90
550
scan

scan就是给一个初始化的数,然后不断的拿前一个结果和最新的值进行处理操作。

let disposeBag = DisposeBag()

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

过滤和约束

filter

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


Observable.of(
    "🐱", "🐰", "🐶",
    "🐸", "🐱", "🐰",
    "🐹", "🐸", "🐱")
    .filter {
        $0 == "🐱"
    }
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

🐱
🐱
🐱

distinctUntilChanged

distinctUntilChanged就是当下一个事件与前一个事件是不同事件的事件才进行处理操作

let disposeBag = DisposeBag()

Observable.of("🐱", "🐷", "🐱", "🐱", "🐱", "🐵", "🐱")
    .distinctUntilChanged()
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)
🐱
🐷
🐱
🐵
🐱
elementAt

只处理在指定位置的事件

let disposeBag = 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

只处理前几个事件信号,

let disposeBag = DisposeBag()

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .take(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)

🐱
🐰
🐶
takeLast

只处理后几个事件信号

let disposeBag = DisposeBag()

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .takeLast(3)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)
🐸
🐷
🐵
takeWhile

当条件满足的时候进行处理

let disposeBag = DisposeBag()

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

1
2
3
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

取消前几个事件

let disposeBag = DisposeBag()

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .skip(2)
    .subscribe(onNext: { print($0) })
    .addDisposableTo(disposeBag)
🐶
🐸
🐷
🐵
skipWhile

满足条件的事件消息都取消

let disposeBag = DisposeBag()

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

4
5
6
skipWhileWithIndex

满足条件的都被取消,传入的闭包同skipWhile有点区别而已

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)

1111

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一个值,然后结束

let disposeBag = DisposeBag()

let sequenceThatFails = 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)
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,968评论 6 482
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,601评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 153,220评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,416评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,425评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,144评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,432评论 3 401
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,088评论 0 261
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,586评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,028评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,137评论 1 334
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,783评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,343评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,333评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,559评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,595评论 2 355
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,901评论 2 345

推荐阅读更多精彩内容