本篇文章,介绍下创建Observable的几种方式,以及如何订阅 销毁Observable
创建Observable
1.just()
- 传入一个默认值
- 指定序列泛型
let observable0 = Observable<Int>.just(5)
//订阅测试
observable0.subscribe {
print("just()_\($0)")
}.disposed(by: disposeBag)
// just()_next(5)
// just()_completed
2.of()
- 传入可变数量的默认值(类型一致)
- 指定序列泛型(可不指定,会自动推断)
let observable1 = Observable<String>.of("A", "B", "C")
//订阅测试
observable1.subscribe {
print("of()_\($0)")
}.disposed(by: disposeBag)
//of()_next(A)
//of()_next(B)
//of()_next(C)
//of()_completed
3.from()
- 指定一个数组参数
let observable2 = Observable.from(["A", "B", "C"])
//订阅测试
observable2.subscribe {
print("from()_\($0)")
}.disposed(by: disposeBag)
//from()_next(A)
//from()_next(B)
//from()_next(C)
//from()_completed
4.empty()
- 创建一个空内容的 Observable 序列
let observable3 = Observable<Int>.empty()
//订阅测试
observable3.subscribe {
print("empty()_\($0)")
}.disposed(by: disposeBag)
//empty()_completed
5.never()
- 创建一个永远不会发出 Event（也不会终止）的 Observable 序列
let observable4 = Observable<Int>.never()
//订阅测试
observable4.subscribe {
print("never()_\($0)")
}.disposed(by: disposeBag)
6.error()
- 直接发送一个错误的 Observable 序列
enum MyError: Error {
case A
case B
}
let observable5 = Observable<Int>.error(MyError.A)
//订阅测试
observable5.subscribe {
print("error()_\($0)")
}.disposed(by: disposeBag)
//error()_error(A)
7.range()
- 指定起始值和结束值,生成从起始值到结束值的序列
let observable6 = Observable.range(start: 1, count: 5)
//订阅测试
observable6.subscribe {
print("range()_\($0)")
}.disposed(by: disposeBag)
//range()_next(1)
//range()_next(2)
//range()_next(3)
//range()_next(4)
//range()_next(5)
//range()_completed
8.repeatElement()
- 创建一个可以无限发出给定元素的 Event的 Observable 序列（永不终止）
let observable7 = Observable.repeatElement(1)
observable7.subscribe {//订阅之后会阻塞UI
print("repeatElement()_\($0)")
}.disposed(by: disposeBag)
///repeatElement()_next(1)
///repeatElement()_next(1)
///repeatElement()_next(1)
///...
9.generate()
- 该方法创建一个只有当提供的所有的判断条件都为 true 的时候，才会给出动作的 Observable 序列。(需要传入: 初始值、迭代、判断条件)
let observable8 = Observable.generate(
initialState: 0,
condition: { $0 <= 10 },
iterate: { $0 + 2 }
)
observable8.subscribe {
print("generate()_\($0)")
}.disposed(by: disposeBag)
//generate()_next(0)
//generate()_next(2)
//generate()_next(4)
//generate()_next(6)
//generate()_next(8)
//generate()_next(10)
//generate()_completed
10.create()
- 该方法接受一个 block 形式的参数，任务是对每一个过来的订阅者进行处理
//这个block有一个回调参数observer就是订阅这个Observable对象的订阅者
//当一个订阅者订阅这个Observable对象的时候，就会将订阅者作为参数传入这个block来执行一些内容
let observable9 = Observable<String>.create{ observer in
//对订阅者发出了.next事件，且携带了一个数据"啊哈哈"
observer.onNext("啊哈哈")
//对订阅者发出了.completed事件
observer.onCompleted()
//因为一个订阅行为会有一个Disposable类型的返回值，所以在结尾一定要returen一个Disposable
return Disposables.create()
}
observable9.subscribe {
print("create()_\($0)")
}.disposed(by: disposeBag)
//create()_next(啊哈哈)
//create()_completed
11.deferred()
- 这个方法相当于是创建一个 Observable 工厂，通过传入一个 block 来执行延迟 Observable序列创建的行为，而这个 block 里就是真正的实例化序列对象的地方
//用于标记是奇数、还是偶数
var isOdd = true
//使用deferred()方法延迟Observable序列的初始化，通过传入的block来实现Observable序列的初始化并且返回。
let factory : Observable<Int> = Observable.deferred {
//让每次执行这个block的时候都会让奇、偶数进行交替
isOdd = !isOdd
//根据isOdd参数，决定创建并返回的是奇数Observable、还是偶数Observable
if isOdd {
return Observable.of(1, 3, 5 ,7)
}else {
return Observable.of(2, 4, 6, 8)
}
}
//第1次订阅测试
factory.subscribe { event in
print("factory__deferred()__\(isOdd)", event)
}.disposed(by: disposeBag)
//第2次订阅测试
factory.subscribe { event in
print("factory__deferred()__\(isOdd)", event)
}.disposed(by: disposeBag)
//factory__deferred()__false next(2)
//factory__deferred()__false next(4)
//factory__deferred()__false next(6)
//factory__deferred()__false next(8)
//factory__deferred()__false completed
//factory__deferred()__true next(1)
//factory__deferred()__true next(3)
//factory__deferred()__true next(5)
//factory__deferred()__true next(7)
//factory__deferred()__true completed
12.interval()
- 创建的 Observable 序列每隔一段设定的时间，会发出一个索引数的元素。而且它会一直发送下去。
(下面方法让其每 1 秒发送一次，并且是在主线程（MainScheduler）发送。)
let observable11 = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
observable11.subscribe { event in
print("interval()__\(event)")
}.disposed(by: disposeBag)
//interval()__next(0)
//interval()__next(1)
//interval()__next(2)
//interval()__next(3)
//interval()__next(4)
// ... 每一秒打印一次, 会一直执行下去
///为了防止会一直执行下去,我们在5秒后,将这个RxSwift.Disposable dispose
let subInterval = observable11.subscribe { event in
print("interval()__\(event)")
}
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 5.0) {
subInterval.dispose()
}
//interval()__next(0)
//interval()__next(1)
//interval()__next(2)
//interval()__next(3)
//interval()__next(4)
13.timer()
- 创建的 Observable序列在经过设定的一段时间后，产生唯一的一个元素。
///5秒后发送一次
let observable12 = Observable<Int>.timer(1, scheduler: MainScheduler.instance)
observable12.subscribe { event in
print("timer()__\(event)")
}.disposed(by: disposeBag)
//timer()__next(0)
//timer()__completed
- 创建的 Observable 序列在经过设定的一段时间后，每隔一段时间产生一个元素。
/// 5秒后,每一秒产生一个元素
let observable13 = Observable<Int>.timer(5, period: 1, scheduler: MainScheduler.instance)
observable13.subscribe(onNext: { (event) in
print("timer()__\(event)")
}, onError: { (error) in
print("timer()__\(error)")
}, onCompleted: {
print("timer()__onCompleted")
}) {
print("timer()__onDisposed")
}
//timer()__0
//timer()__1
//timer()__2
//timer()__3
//timer()__4
//timer()__5
//timer()__6
//timer()__7
//timer()__8
//timer()__9
//...
//为了防止会一直执行下去,我们在10秒后,将这个RxSwift.Disposable dispose
let subscription = observable13.subscribe(onNext: { (event) in
print("timer()__\(event)")
}, onError: { (error) in
print("timer()__\(error)")
}, onCompleted: {
print("timer()__onCompleted")
}) {
print("timer()__onDisposed")
}
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 10.0) {
subscription.dispose()
}
//timer()__0
//timer()__1
//timer()__2
//timer()__3
//timer()__4
//timer()__5
//timer()__onDisposed
以上,就是创建Observable的各种方式.
subscribe --订阅Observable
通过上述测试可知, 我们都是使用 subscribe() 方法来订阅Observable，接收它发出的 Event.
- 使用 subscribe() 订阅了一个Observable 对象，该方法的 block 的回调参数就是被发出的 event 事件
- 初始化 Observable 序列时设置的默认值都按顺序通过 .next 事件发送出来。
- 当 Observable 序列的初始数据都发送完毕，它还会自动发一个 .completed 事件出来。
let observable0 = Observable<Int>.just(5)
observable0.subscribe {
print("just()_\($0)")
}
//just()_next(5)
//just()_completed
2.使用 subscribe() 订阅了一个Observable 对象,通过不同的 block 回调处理不同类型的 event.(onNext、onError、onCompleted 和 onDisposed 这四个回调 block 参数都是有默认值的，即它们都是可选的。所以我们也可以只处理 onNext而不管其他的情况)
let observable0 = Observable<Int>.just(5)
observable0.subscribe { (event) in
print("just()__\(event)")
} onError: { (error) in
print("just()__error")
} onCompleted: {
print("just()__onCompleted")
} onDisposed: {
print("just()__onDisposed")
}
//just()__5
//just()__onCompleted
//just()__onDisposed
监听事件的生命周期 doOn()
- 我们可以使用 doOn 方法来监听事件的生命周期，它会在每一次事件发送前被调用。
- 同时它和 subscribe 一样，可以通过不同的block 回调处理不同类型的 event
通过例子可以看出,
- doOn的onNext onCompleted比subscribe的onNext onCompleted先执行;
- subscribe的onDisposed比doOn的onDisposed先执行
let observable1 = Observable<String>.of("A", "B", "C")
observable1.do(onNext: { (ele) in
print("doOn---\(ele)")
}, onError: { (error) in
print("doOn---\(error)")
}, onCompleted: {
print("doOn---onCompleted")
}, onSubscribe: {
print("doOn---onSubscribe")
}, onSubscribed: {
print("doOn---onSubscribed")
}, onDispose: {
print("doOn---onDispose")
}).subscribe { (event) in
print("subscribe---\(event)")
} onError: { (error) in
print("subscribe---\(error)")
} onCompleted: {
print("subscribe---onCompleted")
} onDisposed: {
print("subscribe---onDisposed")
}.disposed(by: disposeBag)
//doOn---onSubscribe
//doOn---onSubscribed
//doOn---A
//subscribe---A
//doOn---B
//subscribe---B
//doOn---C
//subscribe---C
//doOn---onCompleted
//subscribe---onCompleted
//subscribe---onDisposed
//doOn---onDispose
Observable 的销毁（Dispose）
- 一个 Observable 序列被创建出来后它不会马上就开始被激活从而发出 Event，而是要等到它被某个人订阅了才会激活它。
- 而 Observable 序列激活之后要一直等到它发出了.error或者 .completed的 event 后，它才被终结。
1.dispose()
- 使用该方法我们可以手动取消一个订阅行为.(例子可参考上面interval() timer() 初始化方式中, 取消订阅的行为)
2.DisposeBag
DisposeBag 的对象来管理多个订阅行为的销毁：
我们可以把一个 DisposeBag对象看成一个垃圾袋，把用过的订阅行为都放进去。
而这个DisposeBag 就会在自己快要dealloc 的时候，对它里面的所有订阅行为都调用 dispose()方法。(上文中其实我们已经在使用这个垃圾袋了哦~)
private var disposeBag:DisposeBag = DisposeBag()
网友评论