rx-java
ämnen

Syntax

• Ämne <T, R> subject = AsyncSubject.create (); // Standard AsyncSubject
• Ämne <T, R> subject = BehaviorSubject.create (); // Standard BehaviorSubject
• Ämne <T, R> subject = PublishSubject.create (); // Standard PublishSubject
• Ämne <T, R> subject = ReplaySubject.create (); // Standard ReplaySubject
• mySafeSubject = new SerializedSubject (unSafeSubject); // Konvertera ett unsafeSubject till ett safeSubject - vanligtvis för flertrådade ämnen

parametrar

Anmärkningar

Grundläggande ämnen

Ett Subject i RxJava är en klass som är både en Observable och en Observer . Detta innebär i princip att det kan fungera som ett Observable och skicka inmatningar till prenumeranter och som en Observer att få input från en annan observerbar.

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

Ovanstående skriver ut "Hej, världen!" att konsolera med hjälp av Subjects .

Förklaring

1. Den första raden i koden definierar ett nytt Subject av typen PublishSubject

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

2. Den andra raden prenumererar på ämnet och visar Observer beteende.

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

   Detta gör det möjligt för Subject att ta input som en vanlig abonnent

3. Den tredje raden kallar onNext metod för att visa det Observable beteendet.

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

   Detta gör det möjligt för Subject att ge ingångar till alla som prenumererar på den.

typer

Ett Subject (i RxJava) kan vara av någon av dessa fyra typer:

• AsyncSubject
• BehaviorSubject
• PublishSubject
• ReplaySubject

Ett Subject kan också vara av typen SerializedSubject . Denna typ säkerställer att Subject inte bryter mot det observerbara kontraktet (som anger att alla samtal måste vara serialiserade)

Vidare läsning:

• Att använda eller inte använda ämne från Dave Sextons blogg

PublishSubject

PublishSubject avger endast en Observer de objekt som sänds ut från källan Observable efter prenumerationstidpunkten.

Ett enkelt PublishSubject exempel:

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);

Produktion:

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

I exemplet PublishSubject prenumererar ett PublishSubject på en Observable som fungerar som en klocka och avger objekt (Långt) var 500 miljoner sekund. Som man ser i utgången PublishSubject de ventiler som den får från källan ( clock ) till dess abonnenter ( sub1 och sub2 ).

Ett PublishSubject kan börja PublishSubject ut objekt så fort det skapas, utan någon observatör, vilket riskerar att en eller flera objekt går förlorade tills en observatör kan skänka sig.

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

Produktion:

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

Observera att sub1 avger värden från och med 10 . Förseningen på 5 sekunder orsakade förlust av föremål. Dessa kan inte reproduceras. Detta gör i huvudsak PublishSubject ett Hot Observable .

Observera också att om en observatör prenumererar på PublishSubject efter att den har PublishSubject ut n objekt kan dessa n objekt inte reproduceras för den här observatören.

Nedan är PublishSubject för PublishSubject

PublishSubject avger objekt till alla som har prenumererat, när som helst innan onCompleted för källan Observable kallas.

Om källan Observable avslutas med ett fel, kommer PublishSubject inte att PublishSubject ut några objekt till efterföljande observatörer utan kommer bara att passera felmeddelandet från källan observerbar.

Användningsfall
Anta att du vill skapa en applikation som övervakar aktiekurserna för ett visst företag och vidarebefordrar det till alla kunder som begär det.

/* 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();

I ovanstående användningsfall fungerar PublishSubject som en bro för att vidarebefordra värdena från din server till alla klienter som prenumererar på din watcher .

Vidare läsning:

• PublishSubject javadocs
• Blogg av Thomas Nield (Avancerad läsning)