Rxjava操作符之share()

迷南。 2023-07-01 11:59 63阅读 0赞

顾名思义，Rx的share操作符可以让多个Subscriber共享一个Observable发送的数据，举个例来说明：

val observable = Observable.create<Int> {
        println("emmit start")
        thread(true) {
            repeat(5) { it1 ->
                it.onNext(it1)
            }
        }
    }
    
    observable.subscribe {
        println("subscribe1 : $it")
    }
    
    observable.subscribe {
        println("subscribe2 : $it")
    }
    
    // output:
    // =====================
    // emmit start
    // subscribe1 : 0
    // subscribe1 : 1
    // subscribe1 : 2
    // subscribe1 : 3
    // subscribe1 : 4
    // emmit start
    // subscribe2 : 0
    // subscribe2 : 1
    // subscribe2 : 2
    // subscribe2 : 3
    // subscribe2 : 4

当我们为Observable加上share之后

val observable = Observable.create<Int> {
        ...
    }.share()
    
    //  output:
    // ===============
    //  emmit start
    //  subscribe1 : 0
    //  subscribe1 : 1
    //  subscribe1 : 2
    //  subscribe2 : 2
    //  subscribe1 : 3
    //  subscribe2 : 3
    //  subscribe1 : 4
    //  subscribe2 : 4

日志中emit start只出现了一次，说明两个subscriber中的数据来自同一次发射。

## Observable的Cold与Hot ##

*  Cold Observable
    
     *  没有订阅Observable不会发射数据，订阅的同时开始发射数据
     *  对于所有的Subscriber都会重新发射一批全量数据
 *  Hot Observable
    
     *  Observable随时可以发射数据，无论当前是否有订阅者
     *  不会因为有新的Subscriber就重新发送一遍数据（当然这是默认情况，ReplaySubject等例外）
     *  每次发射的数据会被所有Subscriber共享

默认的Observabe是Cold的，没有添加share之前，每当有新的Subscriber订阅是，都会重新发送一边全量数据。但是添加share之后Observable似乎有些符合Hot的特征。看share的实现，果然如猜测的那样

public final Observable<T> share() {
        return publish().refCount();
    }

publish、refCount，这些都是用来进行冷热流转换的操作符。

## publish()与ConnectableObservable ##

我们经常会使用Subject来实现一个Hot的Observable：

val hot: PublishSubject<Int> = PublishSubject.create()
    
    hot.subscribe(subscriber1)
    hot.subscribe(subscriber2)
    ...
    
    repeat(5) {
        hot.onNext(it)//emit 
    }

除了Subject之外，可以通过publish将一个Cold的Observable转换为ConnectableObservable

val observable = Observable.create<Int> {
       println("emmit start")
       repeat(5) { it1 ->
            it.onNext(it1)
      }
    }
    val connectable = cold.publish()
    
    connectable.subscribe{ println("subscribe1 : $it") }
    connectable.subscribe{ println("subscribe2 : $it") }
    
    println("connect start")
    connectable.connect()
    
    // output:
    // =======================
    // connect start
    // emmit start
    // subscribe1 : 0
    // subscribe2 : 0
    // subscribe1 : 1
    // subscribe2 : 1
    // subscribe1 : 2
    // subscribe2 : 2
    // subscribe1 : 3
    // subscribe2 : 3
    // subscribe1 : 4
    // subscribe2 : 4

ConnectableObservable的行为符合Hot的特性：

*  数据的发射不依赖subscribe，只有调用了connect之后才开始发射数据
 *  emmit start的日志只有一次，所有Subscriber共享同一批数据

## refCount() ##

但是Hot Observable的特性为ConnectableObservable也带来了以下副作用：

*  即使没有Subscriber，也会发射数据

val connectable = cold.publish()
    connectable.connect() //即使没有任何subscibe，调用connect后仍然会发射数据
    
    // output:
    // =======================
    // emmit start

*  Observable不会随着Subscriber的dispose而dispose

val subscriberDisp1 = connectable.subscribe{...}
    val subscriberDisp2 = connectable.subscribe{...}
    
    val observableDisp = connectable.connect()
    
    subscriberDisp1.dispose()
    subscriberDisp2.dispose()
    //即使所有的Subscriber都dispose了，让然需要对Observable进行dispose，否则可能会造成泄露
    observableDisp.dispose()

使用refCount可以帮助我们消灭以上副作用，让Observable变得稍微Cold了一些

val cold = connectable.refCount()
    
    cold.subscribe {
        println("subscribe1 : $it")
    }
    
    cold.subscribe {
        println("subscribe2 : $it")
    }
    
    //  output:
    // ===============
    //  emmit start
    //  subscribe1 : 0
    //  subscribe1 : 1
    //  subscribe1 : 2
    //  subscribe2 : 2
    //  subscribe1 : 3
    //  subscribe2 : 3
    //  subscribe1 : 4
    //  subscribe2 : 4

从日志看跟直接添加share()是一样的（废话，share()=publish().refCount()），数据发射不再依赖connect，当有第一个Subscriber订阅时开始发射，之后新加入的Subscriber共享发射的数据，这符合Hot的特征，从日志也可以看出subscriber2是后加入的，所以少收到了几条数据。另外，refCount保持了对所有Subscriber的引用计数，可以感知到所有Subsciber的dispose行为，然后自动Dispose，避免了泄露的隐患。

## 总结 ##

share的本质是通过一系列转换：**Cold -> putlish() -> Hot -> refCount() -> littleCold**，得到一个兼具冷热特性的observable，似的多个Subscriber可以在Observable的一次发射中共享数据，Rx的很多操作符大多是各种基础操作符的组合，熟练掌握Cold和Hot的概念对我们理解这些操作符的特性很有好处。

![format_png][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cDovL3JlYWN0aXZleC5pby9kb2N1bWVudGF0aW9uL29wZXJhdG9ycy9pbWFnZXMvcHVibGlzaFJlZkNvdW50LmMucG5n?x-oss-process=image/format,png