Erstellen Sie eine Observable für eine Ereignisschleife

Question

Ich habe kürzlich etwas über rx-java gelesen. Ich frage mich, ob der Rahmen die Rechnung für ein Kommunikationssystem zwischen Threads passt. Ich arbeite an einem REST-Server, der in Java geschrieben ist. Jedes Mal, wenn eine Ressource PUT/POSTED ist, möchte ich einige Berechnungen mit einem Pool von Worker-Threads durchführen. Dennoch möchte ich die Anfragen gerne überwachen, vielleicht einige Statistiken ausdrucken. Im Wesentlichen möchte ich ein Observable, also kann ich mit den Anträgen in flexibler Weise mit mehreren Observer s umgehen.

Meine Frage ist, wie kann ich eine geeignete Observable erstellen? Die meisten Anleitungen, die ich gesehen habe, befassen sich mit Operationen auf Observablen wie Mapping usw. Oberwerte werden meist aus Sammlungen oder Ganzzahlbereichen erstellt. Auf jeden Fall scheint es unmöglich zu sein, den erstellten Observablen neue Werte zu geben. Offensichtlich ist die einzige Möglichkeit, diese Flexibilität beizubehalten, Observable.create zu verwenden. Dies scheint jedoch eher gering ausgeprägt zu sein. Ich würde eine Liste der Warteschlangen für jeden neuen Teilnehmer implementieren und einen synchronized Push zu jedem einzelnen tun müssen. Ist das wirklich notwendig oder ist sowas schon in rx-java implementiert?

Answer 1

Was Sie suchen, ist ein Subject. Diese fungieren sowohl als Beobachter als auch als Observable. Zum Beispiel wird ein ReplaySubject alle Ereignisse, die an alle Teilnehmer gesendet werden, abspielen.

Subject<String> replaySubject = ReplaySubject.create(); 
replaySubject.subscribe(s -> System.out.println(s)); 

// elsewhere... 

replaySubject.onNext("First"); 
replaySubject.onNext("Second"); 
replaySubject.onComplete();