ReactiveX のstream分岐について | hiloの情報発信局

ReactiveX のStream分岐は2種類ありますよね。
一つは publish、もう一つはshare

どちらもパッと見同じように見えるのだけれども、
実はちょっとした違いがあります。

一言でいうと、分配と複製になります。
ここでつらつらと説明をしてもイマイチ伝わらないかと思うので、
まずは実際のコードを見ていきましょう！

shareは先程言った２つの内、複製に当たるんですが…
イメージ的には、このように分岐をして複製するみたいな感じです。

最初のObservableが複製されて、
複製されたObservableからSubscribeがコールされます。

このとき、Observableが複製されるので
Observable処理は4回呼ばれます…

うーん…
これはどういう使い方するんだろう…

コードと処理結果はこちらになります！

コード

package com.hilo_islay

import io.reactivex.Observable
import io.reactivex.disposables.CompositeDisposable
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit





fun main(args: Array<String>) {
    var shareCount = 0
    val disposable = CompositeDisposable()
    val observable = Observable.create<Int> { val i = ++shareCount; println("Share :$i"); it.onNext(i) }
        .subscribeOn(Schedulers.computation())
    observable.subscribe()

    observable.share().subscribeOn(Schedulers.computation())
        .subscribe { println("Subscribe :$it") }
        .let { disposable.add(it) }

    observable.share().subscribeOn(Schedulers.computation())
        .subscribe { println("Subscribe :$it") }
        .let { disposable.add(it) }

    observable.share().subscribeOn(Schedulers.computation())
        .subscribe { println("Subscribe :$it") }
        .let { disposable.add(it) }



    TimeUnit.SECONDS.sleep(30)

}

package com.hilo_islay

import io.reactivex.Observable

import io.reactivex.disposables.CompositeDisposable

import io.reactivex.schedulers.Schedulers

import java.util.concurrent.TimeUnit

fun main(args: Array<String>) {

var shareCount = 0

val disposable = CompositeDisposable()

val observable = Observable.create<Int> { val i = ++shareCount; println("Share :$i"); it.onNext(i) }

.subscribeOn(Schedulers.computation())

observable.subscribe()

observable.share().subscribeOn(Schedulers.computation())

.subscribe { println("Subscribe :$it") }

.let { disposable.add(it) }

observable.share().subscribeOn(Schedulers.computation())

.subscribe { println("Subscribe :$it") }

.let { disposable.add(it) }

observable.share().subscribeOn(Schedulers.computation())

.subscribe { println("Subscribe :$it") }

.let { disposable.add(it) }

TimeUnit.SECONDS.sleep(30)

}

処理結果

Share :1
Share :2
Subscribe :2
Share :3
Subscribe :3
Share :4
Subscribe :4

Share(Observable)が4回呼ばれて、Subscribeが3回呼ばれていますね。
分岐だけで使うのは楽なんだけども、親から複製されてしまうので使い所が難しいですね；；

publish

publishは完全に分岐です！
同じように、画像から見ていきましょう。

shareとは違い、Observableが一つでSubscribeが作られた分だけ生成されます。
動きとしてはこちらの方が使い勝手がいいのですが、

publishに関しては「connect()」をコールするまでSubscribeはコールされません。
つまり、うまいこと複製をするタイミングを見極める必要がありますね。

コードと処理結果はこちらです。

コード

package com.hilo_islay

import io.reactivex.Observable
import io.reactivex.disposables.CompositeDisposable
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit

fun main(args : Array<String>) {
    var publishCount= 0
    val disposable = CompositeDisposable()
    val pub = Observable.create<Int>{ val i = ++publishCount; println("Publish $i"); it.onNext(i) }
        .subscribeOn(Schedulers.computation())
        .observeOn(Schedulers.newThread())
        .publish()


    pub.subscribe { println("Subscribe :$it") }.let { disposable.add(it) }
    pub.subscribe { println("Subscribe :$it") }.let { disposable.add(it) }
    pub.subscribe { println("Subscribe :$it") }.let { disposable.add(it) }

    pub.connect()

    TimeUnit.SECONDS.sleep(30)
}

package com.hilo_islay

import io.reactivex.Observable

import io.reactivex.disposables.CompositeDisposable

import io.reactivex.schedulers.Schedulers

import java.util.concurrent.TimeUnit

fun main(args : Array<String>) {

var publishCount= 0

val disposable = CompositeDisposable()

val pub = Observable.create<Int>{ val i = ++publishCount; println("Publish $i"); it.onNext(i) }

.subscribeOn(Schedulers.computation())

.observeOn(Schedulers.newThread())

.publish()

pub.subscribe { println("Subscribe :$it") }.let { disposable.add(it) }

pub.connect()

TimeUnit.SECONDS.sleep(30)

}

処理結果

Publish 1
Subscribe :1
Subscribe :1
Subscribe :1

結果を見てもらうとわかるように、shareと違って、
全てPublishのnextで渡された値が表示されています。

分岐をさせるには、こちらを使っていくのが良さそうですね！

分岐自体はshare、publishの2種類がありますが
基本的にpublishの方を使っていけば大丈夫だと思います！！

share

コード

処理結果

publish

コード

処理結果

Leave a Reply Cancel reply