RxScala Subscribe mit multi Observer nur Ereignis an erster Stelle senden

Question

Ich versuche verwenden multi Observer abonnieren Sie eine Observable welche onNext in einer Schleife aufgetreten ist.Es scheint nicht für jeden Beobachter funktionieren.

import rx.lang.scala.Observable 

object SubscribeMultiEvent extends App{ 
    val obv = Observable.apply[String]{ s => 
    def printForever: Unit = { 
     s.onNext("hi~") 
     Thread.sleep(1000) 
     printForever 
    } 
    printForever 
    } 

    obv.subscribe(s => println(s"first observer - $s")) 
    obv.subscribe(s => println(s"second observer - $s")) 

    Thread.currentThread().join() 
} 

Antwort nur für den ersten Beobachter

first observer - hi~ 
first observer - hi~ 
... 

Warum zweiten der Zeichnung nicht empfangen kann? Danke

Answer 1

Das Problem in Ihren Codes ist Ihre Observable ist synchron. Es bedeutet, dass die zweite subscribe nicht läuft, bis die erste subscribe beendet ist. Und da Ihre Observable nie abgeschlossen ist, kann die zweite subscribe nicht ausgeführt werden.

Um dieses Problem zu beheben, müssen Sie Ihre Observable asynchron machen. Sie können subscribeOn verwenden, um in einem anderen Thread zu laufen. Zum Beispiel

import rx.lang.scala.Observable 
import rx.lang.scala.schedulers.NewThreadScheduler 

object SubscribeMultiEvent extends App{ 
    val obv = Observable.apply[String]{ s => 
    def printForever: Unit = { 
     s.onNext("hi~") 
     Thread.sleep(1000) 
     printForever 
    } 
    printForever 
    }.subscribeOn(NewThreadScheduler()) 

    obv.subscribe(s => println(s"first observer - $s")) 
    obv.subscribe(s => println(s"second observer - $s")) 

    Thread.sleep(60000) 
} 

Thread.sleep(60000) am Ende ist wichtig. Die Threads von RxJava sind standardmäßig Daemon, und wenn der Hauptthread beendet ist, weil es keine Nicht-Daemon-Threads mehr gibt, wird JVM beendet. Um zu verhindern, dass der Haupt-Thread anhält, müssen Sie etwas wie Thread.sleep(60000) hinzufügen.