RxSwift的使用详解02

发表于 更新于 分类于 阅读次数： 本文字数： 8.1k 阅读时长 ≈ 7 分钟
  • 上一篇: RxSwift的使用详解01主要介绍了
    • RxSwift简单体验(在控件中的简单使用)
    • RxSwift常见操作(never, just, of, empty, creat等10个sequence的使用)
    • RxSwift中Subjects
    • 变换操作(map, flatMap等)和资源释放DisposeBag
    • UIBindingObserver创建自己的监听者
  • 本文主要内容
    • 联合操作: 把多个Observable流合成单个Observable流
    • elementAt, single, skip等过滤和约束操作
    • toArray, reduce, concat等数学操作

联合操作

  • 联合操作就是把多个Observable流合成单个Observable流

startWith

  • 在发出事件消息之前，先发出某个特定的事件消息。
  • 比如发出事件2 ，3然后我startWith(1)，那么就会先发出1，然后2 ，3.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
//未添加startWith
Observable.of("2", "3").subscribe({ print($0) }).addDisposableTo(bag)
    /***输出顺序:
        next(2)
        next(3)
        completed
    ***/

//使用startWith
Observable.of("2", "3").startWith("1").subscribe({ print($0) }).addDisposableTo(bag)
        
    /*输出顺序:
         next(1)
         next(2)
         next(3)
         completed
     */

merge

  • 合并两个Observable流合成单个Observable流，根据时间轴发出对应的事件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()
Observable.of(subject1, subject2)
    .subscribe({ print($0) })
    .addDisposableTo(bag)
    
print("-------------------------")

Observable.of(subject1, subject2)
    .merge()
    .subscribe({ print($0) })
    .addDisposableTo(bag)
        
subject1.onNext("quan")
subject1.onNext("jun")
subject2.onNext("ya")
subject2.onNext("jie")
subject1.onNext("tian")
subject2.onNext("guo")
        
    /*输出事件:
         next(quan)
         next(jun)
         next(ya)
         next(jie)
         next(tian)
         next(guo)
     */

zip

  • 绑定超过最多不超过8个的Observable流，结合在一起处理。
  • 注意Zip是一个事件对应另一个流一个事件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
let subject3 = PublishSubject<String>()
let subject4 = PublishSubject<String>()
Observable.zip(subject3, subject4) { (sub3, sub4) -> String in
        sub3 + "+" + sub4
    }.subscribe({ print($0) })
    .addDisposableTo(bag)
        
subject3.onNext("quan")
subject3.onNext("jun")
subject4.onNext("ya")
subject4.onNext("jie")
subject3.onNext("tian")
subject4.onNext("guo")
        
    /*输出事件:
         将subject3和subject4压缩到一起共同处理
         next(quan+ya)
         next(jun+jie)
         next(tian+guo)
     */

combineLatest

  • 绑定超过最多不超过8个的Observable流，结合在一起处理。
  • 和Zip不同的是combineLatest是一个流的事件对应另一个流的最新的事件，两个事件都会是最新的事件，可将下图与Zip的图进行对比
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
let subject5 = PublishSubject<String>()
let subject6 = PublishSubject<String>()
Observable.combineLatest(subject5, subject6) { (sub5, sub6) -> String in
        sub5 + "+" + sub6
    }.subscribe({ print($0) }).addDisposableTo(bag)

subject5.onNext("quan")
subject5.onNext("1")
subject6.onNext("ya")
subject6.onNext("2")
subject5.onNext("--")

    /*输出事件:
         将subject3的最新事件和subject4的最新事件一起处理
         next(1+ya)
         next(1+2)
         next(--+2)
     */

switchLatest

  • switchLatest可以对事件流进行转换，本来监听的subject1，我可以通过更改variable里面的value更换事件源。变成监听subject2了
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
let subject7 = BehaviorSubject(value: "love")
//BehaviorSubject: 接受订阅之前的最后一个事件
let subject8 = BehaviorSubject(value: "love to")
let variable = Variable(subject7)
variable.asObservable()
        .switchLatest()
        .subscribe({ print($0) })
        .addDisposableTo(bag)
        
subject7.onNext("ya")
subject7.onNext("jie")

variable.value = subject8
subject7.onNext("quan")
subject8.onNext("jun")
        
variable.value = subject7
subject8.onNext("jie")
subject7.onNext("guo")

    /*输出事件:
         next(love)
         next(ya)
         next(jie)
         next(love to)
         next(jun)
         next(quan)
         next(guo)
     */

过滤和约束

distinctUntilChanged

  • distinctUntilChanged就是当: 下一个事件与前一个事件是不同事件的事件才进行处理操作
1
2
3
4
5
6
7
8
9
10
11
12
13
Observable.of(1, 2, 1, 1, 1, 3, 3, 1)
        .distinctUntilChanged()
        .subscribe({ print($0) })
        .addDisposableTo(bag)
        
    /*输出顺序为:
         next(1)
         next(2)
         next(1)
         next(3)
         next(1)
         completed
     */

elementAt

  • 只处理在指定位置的事件
1
2
3
4
5
6
7
8
Observable.of(1, 2, 3, 4, 5)
        .elementAt(3)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(4)
         completed
     */

single

  • 找出在sequence只发出一次的事件，如果超过一个就会发出error错误

多个信号输出的情况

1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4)
        .single()
        .subscribe({ print($0) })
        .addDisposableTo(bag)
        
    /*输出顺序为:
         next(1) //单一信号超过了一个,只会输出第一个,然后输出error
         error(Sequence contains more than one element.)
     */

指定某唯一信号的情况

1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4)
        .single({ $0 == 2 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(2)
         completed
     */

指定某不唯一信号的情况

1
2
3
4
5
6
7
8
9
10

Observable.of(1, 4, 3, 4)
        .single({ $0 == 4 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    
    /*输出顺序为:
         next(4) //单一信号超过了一个,只会输出第一个,然后输出error
         error(Sequence contains more than one element.)
     */

找不到该信号的情况

1
2
3
4
5
6
7
8
Observable.of(1, 4, 3, 4)
        .single({ $0 == 2 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         没有对应的参数,然后输出error
         error(Sequence doesn't contain any elements.)
     */

filter

  • 过滤掉某些不符合要求的事件
1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4, 5)
        .filter({ $0 > 3 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

take

  • 只处理前几个事件信号
1
2
3
4
5
6
7
8
9
10

Observable.of(1, 2, 3, 4, 5)
        .take(2)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(1)
         next(2)
         completed
     */

takeLast

  • 只处理后几个事件信号
1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4, 5)
        .takeLast(2)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

takeWhile

  • 当条件满足的时候进行处理
1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4, 5)
        .takeWhile({ $0 > 3 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

takeUntil

  • 接收事件消息，直到另一个sequence发出事件消息的时候.停止接收消息,输出completed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()
subject1.takeUntil(subject2)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
subject1.onNext("quan")
subject1.onNext("jun")

subject2.onNext("ya")//停止接收消息
        
subject1.onNext("tian")
subject2.onNext("guo")

    /*输出顺序为:
         next(quan)
         next(jun)
         completed
     */

skip

  • 取消前几个事件
1
2
3
4
5
6
7
8
9
10
Observable.of(1, 2, 3, 4, 5)
        .skip(3)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
        
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

skipWhile

  • 满足条件的事件消息都取消
1
2
3
4
5
6
7
8
9
Observable.of(1, 2, 3, 4, 5)
        .skipWhile({ $0 < 4 })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

skipWhileWithIndex

  • 满足条件的都被取消，传入的闭包同skipWhile有点区别而已
  • skipWhile的(<4)和skipWhileWithIndex的(<=3)的效果是一样的
1
2
3
4
5
6
7
8
9
10
11
12
Observable.of(1, 2, 3, 4, 5)
        .skipWhileWithIndex({ (element, index) -> Bool in
            index <= 3
        })
        .subscribe({ print($0) })
        .addDisposableTo(bag)
    
    /*输出顺序为:
         next(4)
         next(5)
         completed
     */

skipUntil

  • 直到某个sequence发出了事件消息，才开始接收当前sequence发出的事件消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
let subject3 = PublishSubject<String>()
let subject4 = PublishSubject<String>()
subject3.skipUntil(subject4)
        .subscribe({ print($0) })
        .addDisposableTo(bag)
subject3.onNext("quan")
subject4.onNext("jun")
        
subject4.onNext("ya")//开始接收消息
        
subject3.onNext("tian")
subject4.onNext("guo")
    
    /*输出顺序为:
         next(tian)
     */

数学操作

toArray

  • 将sequence转换成一个array，并转换成单一事件信号，然后结束
1
2
3
4
5
6
7
8
9
Observable.range(start: 1, count: 5)
        .toArray()
        .subscribe({ print($0) })
        .addDisposableTo(bag)
        
    /*输出顺序为:
         next([1, 2, 3, 4, 5])
         completed
     */

reduce

  • 用一个初始值，对事件数据进行累计操作。reduce接受一个初始值，和一个操作符号
1
2
3
4
5
6
7
8
9
Observable.of(10, 12, 34)
       .reduce(0, accumulator: +)
       .subscribe({ print($0) })
       .addDisposableTo(bag)
       
  /*输出顺序为:
         next(56)
         completed
     */

concat

  • concat会把多个sequence和并为一个sequence，并且当前面一个sequence发出了completed事件，才会开始下一个sequence的事件。
  • 在第一sequence发出onCompleted完成之前，第二个sequence发出的事件都会被忽略
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
let subject1 = BehaviorSubject(value: "quan")
let subject2 = BehaviorSubject(value: "jun")
let variable = Variable(subject1)
variable.asObservable()
        .concat()
        .subscribe({ print($0) })
        .addDisposableTo(bag)
subject1.onNext("ya")
subject1.onNext("jie")
subject2.onNext("jun")  //subject2不被输出
        
variable.value = subject2  //subject1发出onCompleted()之前会继续输出subject1
        
subject1.onNext("guo")
subject2.onNext("tian")
        
subject1.onCompleted()  //subject1结束,开始输出subject2,此时subject2的值接受最后一个("tian")
        
subject2.onNext("love")
subject1.onNext("to love") //subject1将不再被输出
        
    /*输出顺序为:
        next(quan)
        next(ya)
        next(jie)
        next(guo)
        next(tian)
        next(love)
     */

RxSwift的优点

  • Composable 可组合，在设计模式中有一种模式叫做组合模式，你可以方便的用不同的组合实现不同的类
  • Reusable 代码可重用，原因很简单，对应RxSwift，就是一堆Obserable
  • Declarative 响应式的，因为状态不可变，只有数据变化
  • Understandable and concise 简洁，容易理解。
  • Stable 稳定，因为RxSwift写出的代码，单元测试时分方便
  • Less stateful “无”状态性，因为对于响应式编程，你的应用程序就是一堆数据流
  • Without leaks 没有泄漏，因为资源管理非常简单