rx-java
Themen

Syntax

• Betreff <T, R> subject = AsyncSubject.create (); // Standard AsyncSubject
• Betreff <T, R> subject = BehaviorSubject.create (); // Standard BehaviorSubject
• Betreff <T, R> subject = PublishSubject.create (); // Default PublishSubject
• Betreff <T, R> subject = ReplaySubject.create (); // Standard ReplaySubject
• mySafeSubject = neues SerializedSubject (unSafeSubject); // Konvertieren Sie ein unsafeSubject in ein safeSubject - im Allgemeinen für Betreffs mit mehreren Threads

Parameter

Bemerkungen

Grundthemen

Ein Subject in RxJava ist eine Klasse , die sowohl eine ist Observable und einen Observer . Dies bedeutet im Wesentlichen, dass es als Observable fungieren und Eingaben an Abonnenten und als Observer , um Eingaben von einem anderen Observable zu erhalten.

Subject<String, String> subject = PublishSubject.create(); 
subject.subscribe(System.out::print);
subject.onNext("Hello, World!"); 

Die obigen Drucke "Hallo, Welt!" zu konsultieren mit Subjects .

Erläuterung

1. Die erste Codezeile definiert einen neuen Subject des Typs PublishSubject

   Subject<String, String> subject = PublishSubject.create();
       |     |       |       |                 |
    subject<input, output>  name   = default publish subject
   

2. Die zweite Zeile abonniert das Thema und zeigt das Verhalten des Observer .

   subject.subscribe(System.out::print);
   

   Dies ermöglicht dem Subject , Eingaben wie ein normaler Teilnehmer zu machen

3. Die dritte Zeile ruft die onNext Methode des Subjekts, das zeigt Observable Verhalten.

   subject.onNext("Hello, World!"); 
   

   Dies ermöglicht dem Subject , Eingaben für alle Abonnenten zu geben.

Typen

Ein Subject (in RxJava) kann einen der folgenden vier Typen haben:

• AsyncSubject
• BehaviorSubject
• PublishSubject
• ReplaySubject

Ein Subject kann auch vom Typ SerializedSubject . Dieser Typ stellt sicher, dass der Subject nicht gegen den beobachtbaren Vertrag verstößt (der besagt, dass alle Anrufe serialisiert werden müssen).

Lesen Sie weiter:

• Betreff aus Dave Sextons Blog verwenden oder nicht verwenden

PublishSubject

PublishSubject gibt nur diejenigen Elemente an einen Observer , die nach dem Abonnement von der Quelle Observable PublishSubject .

Ein einfaches PublishSubject Beispiel:

Observable<Long> clock = Observable.interval(500, TimeUnit.MILLISECONDS);
Subject<Long, Long> subjectLong = PublishSubject.create();

clock.subscribe(subjectLong);

System.out.println("sub1 subscribing...");
subjectLong.subscribe(l -> System.out.println("sub1 -> " + l));
Thread.sleep(3000);
System.out.println("sub2 subscribing...");
subjectLong.subscribe(l -> System.out.println("sub2 -> " + l));
Thread.sleep(5000);

Ausgabe:

sub1 subscribing...
sub1 -> 0
sub1 -> 1
sub2 subscribing...
sub1 -> 2
sub2 -> 2
sub1 -> 3
sub2 -> 3

Im obigen Beispiel abonniert ein PublishSubject ein Observable das sich wie eine Uhr PublishSubject und alle 500 PublishSubject Elemente (Long) ausgibt. Wie in der Ausgabe zu sehen ist, gibt PublishSubject die PublishSubject weiter, die es von der Quelle ( clock ) an seine Subskribenten ( sub1 und sub2 ) sub2 .

Ein PublishSubject kann mit dem Ausgeben von Objekten beginnen, sobald es erstellt wurde, ohne dass ein Beobachter die Gefahr besteht, dass ein oder mehrere Objekte verloren gehen, bis ein Beobachter sich sonnen kann.

createClock(); // 3 lines moved for brevity. same as above example

Thread.sleep(5000); // introduces a delay before first subscribe

sub1andsub2(); // 6 lines moved for brevity. same as above example

Ausgabe:

sub1 subscribing...
sub1 -> 10
sub1 -> 11
sub2 subscribing...
sub1 -> 12
sub2 -> 12
sub1 -> 13
sub2 -> 13

Beachten Sie, dass sub1 Werte ab 10 sub1 . Die Verzögerung von 5 Sekunden führte zum Verlust von Artikeln. Diese können nicht reproduziert werden. Dies macht PublishSubject Wesentlichen zu einem Hot Observable .

Beachten Sie auch, dass, wenn ein Beobachter das PublishSubject abonniert, nachdem n Elemente PublishSubject wurden, diese n Elemente für diesen Beobachter nicht reproduziert werden können .

Unten sehen Sie das PublishSubject von PublishSubject

PublishSubject gibt Elemente an alle Abonnenten aus, die zu einem beliebigen Zeitpunkt vor dem onCompleted von onCompleted der Quelle Observable aufgerufen wurden.

Wenn die Quelle Observable mit einem Fehler PublishSubject , PublishSubject keine Elemente an nachfolgende Beobachter, sondern leitet einfach die Fehlerbenachrichtigung von der Quelle Observable weiter.

Anwendungsfall
Angenommen, Sie möchten eine Anwendung erstellen, die die Aktienkurse eines bestimmten Unternehmens überwacht und an alle Kunden weiterleitet, die dies verlangen.

/* Dummy stock prices */
Observable<Integer> prices = Observable.just(11, 12, 14, 11, 10, 12, 15, 11, 10);

/* Your server */
PublishSubject<Integer> watcher = PublishSubject.create();
/* subscribe to listen to stock price changes and push to observers/clients */
prices.subscribe(watcher);

/* Client application */
stockWatcher = getWatcherInstance(); // gets subject
Subscription steve = stockWatcher.subscribe(i -> System.out.println("steve watching " + i));
Thread.sleep(1000);
System.out.println("steve stops watching");
steve.unsubscribe();

Im obigen Anwendungsbeispiel dient PublishSubject als Brücke, um die Werte von Ihrem Server an alle Clients weiterzuleiten, die Ihren watcher abonnieren.

Lesen Sie weiter:

• PublishSubject- Javadocs
• Blog von Thomas Nield (Fortgeschrittene Lektüre)