RxSwift学习之旅 - share vs replay vs shareReplay

发表于   |   分类于   |  

抛出问题

开发中最容易犯的错误是忘了每个订阅者都会导致observable重新执行链式调用。

like this:

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
31
32
33
func request() -> Observable<String>{
        return Observable.create{
            observer in
            print("发送网络请求")
            observer.onNext("请求成功!")
            return Disposables.create {
                
            }
        }
    }
    
let results = query.rx.text
            .orEmpty
            .asObservable()
            .filter{
                $0.characters.count > 0
            }
            .flatMapLatest{
                query in
                self.request()
        }
        
        results.subscribe{
                print("订阅者 one: \($0)")
            }.disposed(by: disposeBag)
        
        results.subscribe{
                print("订阅者 two: \($0)")
            }.disposed(by: disposeBag)
        
        results.subscribe{
                print("订阅者 three: \($0)")
            }.disposed(by: disposeBag)

三个订阅者会发送三次请求。

解决？

当有多个订阅者去订阅同一个Observable的时候，我们不希望Observable每次有新的订阅者都去执行。

RxSwift提供了很多操作:

share()、replay()、replayAll()、shareReplay()、publish()、shareReplayLatesWhileConnected()， 这么多个，应该选哪一个？

你现在能够说出他们之间的不同吗？

先看下总体的比较，有个大概的了解:

image

1 表示重播最多bufferSize个事件
2 表示当订阅者的引用计数大于0时，重播一个事件

共享订阅者

多个订阅者订共享一个订阅者对象

可连接

可连接序列只有调用connect后才会开始发射值，可以等多个订阅者订阅后再连接。

引用计数

返回的observable记录了订阅者的数量，当订阅者数量从0变成1，订阅源序列，当订阅者数量从1变成0，取消订阅并重置源序列。

每次订阅者数量从0变成1源序列将会重新被订阅。

重播事件

重播已经发射的事件给订阅者。

replay(bufferSize)和shareReplay(bufferSize)最多重播bufferSize个，而shareReplayLatestWhileConnected最多一个，当订阅者的引用计数变成0，buffer会被清空，所以引用计数从0变成1，订阅者不会受到重播事件。

publish

以下例子使用interval模拟用户输入文本，并进行搜索:

1
2
3
4
5
6
7
8
9
var results:Observable<String>!
results = Observable<Int>
            .interval(1, scheduler: MainScheduler.instance)
            .map{
                "\($0)"
            }
            .flatMapLatest{
                query in
                self.request(query)

然后去订阅结果:

publish原来讲到过，只有connect之后才会发射值。

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
func publish(){
    let results = self.results.publish()
    
    results.subscribe{
            print("订阅者 one: \($0)")
        }.disposed(by: disposeBag)
    
    results.subscribe{
            print("订阅者 two: \($0)")
        }.disposed(by: disposeBag)
    
    _ = results.connect()
    
    delay(4){
        print("three 订阅")
        results.subscribe{
            print("订阅者 three: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
订阅者 two: next(3 请求成功!)
three 订阅                        //没有重播事件
搜索 4  发送网络请求
订阅者 one: next(4 请求成功!)
订阅者 two: next(4 请求成功!)
订阅者 three: next(4 请求成功!)

replayAll

重播所有事件:

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
31
32
33
34
35
func replayAll(){
    
    let results = self.results.replayAll()
    
    results.subscribe{
        print("订阅者 one: \($0)")
        }.disposed(by: disposeBag)
    
    results.subscribe{
        print("订阅者 two: \($0)")
        }.disposed(by: disposeBag)
    
    _ = results.connect()
    
    delay(4){
        print("three 订阅")
        results.subscribe{
            print("订阅者 three: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
订阅者 two: next(3 请求成功!)
three 订阅
订阅者 three: next(0 请求成功!)     //订阅后，受到重播的所有事件
订阅者 three: next(1 请求成功!)
订阅者 three: next(2 请求成功!)
订阅者 three: next(3 请求成功!)
搜索 4  发送网络请求
订阅者 one: next(4 请求成功!)
订阅者 two: next(4 请求成功!)
订阅者 three: next(4 请求成功!)

replay

replay(bufferSize)重播指定个数的事件:

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
31
32
func replay(){
    let results = self.results.replay(2)
    
    results.subscribe{
        print("订阅者 one: \($0)")
        }.disposed(by: disposeBag)
    
    results.subscribe{
        print("订阅者 two: \($0)")
        }.disposed(by: disposeBag)
    
    _ = results.connect()
    
    delay(4){
        print("three 订阅")
        results.subscribe{
            print("订阅者 three: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
订阅者 two: next(3 请求成功!)
three 订阅                        //重播最后2个事件
订阅者 three: next(2 请求成功!)
订阅者 three: next(3 请求成功!)
搜索 4  发送网络请求
订阅者 one: next(4 请求成功!)
订阅者 two: next(4 请求成功!)
订阅者 three: next(4 请求成功!)

share

订阅者从1变成0重置序列:

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
func share(){
    let results = self.results.share()
    
    let sub = results.subscribe{
        
            print("订阅者 one: \($0)")
        }
    
    delay(4){
        //订阅者从1变成0
        //可被观察序列重新发射
        print("订阅者 one被销毁")
        
        sub.dispose()
        
        results.subscribe{
            print("订阅者 two: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
订阅者 one被销毁
搜索 0  发送网络请求                //从0开始重新发射
订阅者 two: next(0 请求成功!)

shareReplay(bufferSize)

在share的基础重播bufferSize个值。

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
31
func shareReplay(){
    let results = self.results.shareReplay(2)
    
    let sub1 = results.subscribe{
        print("订阅者 one: \($0)")
        }
    
    let sub2 = results.subscribe{
        print("订阅者 two: \($0)")
        }
    
    delay(4){
        sub1.dispose()
        sub2.dispose()
        
        print("three 订阅")
        results.subscribe{
            print("订阅者 three: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
订阅者 two: next(3 请求成功!)
three 订阅
订阅者 three: next(2 请求成功!)         //虽然订阅者都被销毁，但是仍收到最后两个值
订阅者 three: next(3 请求成功!)
搜索 0  发送网络请求
订阅者 three: next(0 请求成功!)

shareReplayLatestWhileConnected

订阅者从1变成0，缓存区被清空:

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
func shareReplayLatestWhileConnected(){
    let results = self.results.shareReplayLatestWhileConnected() // test  shareReplay(1)
    
    let sub = results.subscribe{
                print("订阅者 one: \($0)")
            }
    
    delay(4){
        sub.dispose()
        //订阅者从1变成0， 缓存的一个值被清掉了，所以不会收到最后一个值
        print("two 订阅")
        results.subscribe{
            print("订阅者 two: \($0)")
        }.disposed(by: self.disposeBag)
    }
}

output:
搜索 3  发送网络请求
订阅者 one: next(3 请求成功!)
two 订阅
搜索 0  发送网络请求
订阅者 two: next(0 请求成功!)
搜索 1  发送网络请求
订阅者 two: next(1 请求成功!)

试试改成shareReplay(1)，或者不销毁sub。

总结

可以结合上面的例子，在实际运用中选择合适的接口。

代码见github:

RxSwiftShareOrReplay

AloneMonkey wechat
欢迎您扫一扫上面的微信公众号，订阅我的博客！