RxJava - Guida rapida

RxJava è un'estensione basata su Java di ReactiveX. Fornisce l'implementazione o il progetto ReactiveX in Java. Di seguito sono riportate le caratteristiche chiave di RxJava.

• Estende il pattern dell'osservatore.

• Supporta sequenze di dati / eventi.

• Fornisce operatori per comporre sequenze insieme in modo dichiarativo.

• Gestisce internamente il threading, la sincronizzazione, la sicurezza dei thread e le strutture dati concorrenti.

Cos'è ReactiveX?

ReactiveX è un progetto che mira a fornire un concetto di programmazione reattiva a vari linguaggi di programmazione. La programmazione reattiva si riferisce allo scenario in cui il programma reagisce quando e quando vengono visualizzati i dati. È un concetto di programmazione basato su eventi e gli eventi possono propagarsi agli osservatori dei registri.

Secondo il Reactive, hanno combinato il meglio del pattern Observer, del pattern Iterator e del pattern funzionale.

Il modello Observer è stato eseguito correttamente. ReactiveX è una combinazione delle migliori idee del pattern Observer, del pattern Iterator e della programmazione funzionale.

Programmazione funzionale

La programmazione funzionale ruota attorno alla creazione del software utilizzando funzioni pure. Una funzione pura non dipende dallo stato precedente e restituisce sempre lo stesso risultato per gli stessi parametri passati. Pure functions aiuta a evitare problemi associati a oggetti condivisi, dati mutabili ed effetti collaterali spesso prevalenti negli ambienti multi-threading.

Programmazione reattiva

La programmazione reattiva si riferisce alla programmazione guidata dagli eventi in cui i flussi di dati arrivano in modo asincrono e vengono elaborati quando arrivano.

Programmazione funzionale reattiva

RxJava implementa entrambi i concetti insieme, in cui i dati dei flussi cambiano nel tempo e la funzione del consumatore reagisce di conseguenza.

Il Manifesto Reattivo

Reactive Manifesto è un documento in linea che afferma l'elevato standard dei sistemi software applicativi. Come da manifesto, di seguito sono riportati gli attributi chiave di un software reattivo:

• Responsive - Dovrebbe sempre rispondere in modo tempestivo.

• Message Driven - Utilizzare il passaggio di messaggi asincrono tra i componenti in modo che mantengano l'accoppiamento libero.

• Elastic - Dovrebbe rimanere reattivo anche sotto carico elevato.

• Resilient - Dovrebbe rimanere reattivo anche in caso di guasto di uno o più componenti.

Componenti chiave di RxJava

RxJava ha due componenti chiave: Observables e Observer.

• Observable - Rappresenta un oggetto simile a Stream che può emettere zero o più dati, può inviare messaggi di errore, la cui velocità può essere controllata durante l'emissione di un insieme di dati, può inviare dati sia finiti che infiniti.

• Observer- Sottoscrive i dati di sequenza di Observable e reagisce per elemento degli osservabili. Gli osservatori vengono informati ogni volta che Observable emette un dato. Un Observer gestisce i dati uno per uno.

Un osservatore non viene mai avvisato se gli elementi non sono presenti o se non viene restituita una richiamata per un elemento precedente.

Configurazione dell'ambiente locale

RxJava è una libreria per Java, quindi il primo requisito è avere JDK installato nella tua macchina.

Requisito del sistema

JDK	1.5 o superiore.
Memoria	Nessun requisito minimo.
Spazio sul disco	Nessun requisito minimo.
Sistema operativo	Nessun requisito minimo.

Passaggio 1: verifica dell'installazione di Java sulla macchina

Prima di tutto, apri la console ed esegui un comando java basato sul sistema operativo su cui stai lavorando.

OS	Compito	Comando
finestre	Apri la Console di comando	c: \> java -version
Linux	Apri Terminale di comando	$ java -version
Mac	Apri Terminale	macchina: <joseph $ java -version

Verifichiamo l'output per tutti i sistemi operativi -

OS	Produzione
finestre	versione java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)
Linux	versione java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)
Mac	versione java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)

Se non hai Java installato sul tuo sistema, scarica il Java Software Development Kit (SDK) dal seguente collegamento https://www.oracle.com. Stiamo assumendo Java 1.8.0_101 come versione installata per questo tutorial.

Passaggio 2: impostare l'ambiente JAVA

Impostare il JAVA_HOMEvariabile di ambiente in modo che punti alla posizione della directory di base in cui Java è installato sulla macchina. Per esempio.

OS	Produzione
finestre	Imposta la variabile d'ambiente JAVA_HOME su C: \ Program Files \ Java \ jdk1.8.0_101
Linux	export JAVA_HOME = / usr / local / java-current
Mac	export JAVA_HOME = / Library / Java / Home

Aggiungi la posizione del compilatore Java al percorso di sistema.

OS	Produzione
finestre	Aggiungi la stringa C:\Program Files\Java\jdk1.8.0_101\bin alla fine della variabile di sistema, Path.
Linux	export PATH = $ PATH: $ JAVA_HOME / bin /
Mac	non richiesto

Verifica l'installazione di Java utilizzando il comando java -version come spiegato sopra.

Passaggio 3: scarica l'archivio RxJava2

Scarica l'ultima versione del file jar RxJava da RxJava @ MVNRepository e la sua dipendenza Reactive Streams @ MVNRepository . Al momento della stesura di questo tutorial, abbiamo scaricato rxjava-2.2.4.jar, reactive-streams-1.0.2.jar e li abbiamo copiati nella cartella C: \> RxJava.

OS	Nome dell'archivio
finestre	rxjava-2.2.4.jar, reattive-stream-1.0.2.jar
Linux	rxjava-2.2.4.jar, reattive-stream-1.0.2.jar
Mac	rxjava-2.2.4.jar, reattive-stream-1.0.2.jar

Passaggio 4: impostare l'ambiente RxJava

Impostare il RX_JAVAvariabile di ambiente in modo che punti alla posizione della directory di base in cui il jar RxJava è memorizzato sulla macchina. Supponiamo di aver archiviato rxjava-2.2.4.jar e reactive-streams-1.0.2.jar nella cartella RxJava.

Suor n	Sistema operativo e descrizione
1	Windows Imposta la variabile d'ambiente RX_JAVA su C: \ RxJava
2	Linux esporta RX_JAVA = / usr / local / RxJava
3	Mac esporta RX_JAVA = / Library / RxJava

Passaggio 5: impostare la variabile CLASSPATH

Impostare il CLASSPATH variabile di ambiente in modo che punti alla posizione del jar RxJava.

Suor n	Sistema operativo e descrizione
1	Windows Imposta la variabile d'ambiente CLASSPATH su% CLASSPATH%;% RX_JAVA% \ rxjava-2.2.4.jar;% RX_JAVA% \ reattive-streams-1.0.2.jar;.;
2	Linux export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: reattive-flussi-1.0.2.jar :.
3	Mac export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: reattive-flussi-1.0.2.jar :.

Passaggio 6: testare l'installazione di RxJava

Crea una classe TestRx.java come mostrato di seguito -

import io.reactivex.Flowable;
public class TestRx {
   public static void main(String[] args) {
      Flowable.just("Hello World!")
         .subscribe(System.out::println);
   }
}

Passaggio 7: verifica del risultato

Compila le classi usando javac compilatore come segue -

C:\RxJava>javac Tester.java

Verifica l'output.

Hello World!

Observables rappresenta le fonti di dati dove come Observers (Subscribers)ascoltali. In poche parole, un osservabile emette oggetti e un abbonato li consuma.

Osservabile

• Observable fornisce i dati una volta che l'abbonato inizia ad ascoltare.

• Observable può emettere un numero qualsiasi di elementi.

• Anche Observable può emettere solo un segnale di completamento senza alcun elemento.

• Observable può terminare con successo.

• Observable potrebbe non terminare mai. ad esempio, è possibile fare clic su un pulsante un numero qualsiasi di volte.

• Observable può generare errori in qualsiasi momento.

Abbonato

• Observable può avere più iscritti.

• Quando un Observable emette un oggetto, ogni metodo onNext () dell'abbonato viene richiamato.

• Quando un Observable ha terminato di emettere elementi, ogni metodo onComplete () dell'abbonato viene richiamato.

• Se un Observable genera un errore, viene richiamato ogni metodo onError () dell'abbonato.

Di seguito sono riportate le classi base per creare osservabili.

• Flowable- 0..N flussi, emette 0 o n elementi. Supporta flussi reattivi e contropressione.

• Observable - 0..N flussi, ma nessuna contropressione.

• Single- 1 articolo o errore. Può essere trattato come una versione reattiva della chiamata al metodo.

• Completable- Nessun elemento emesso. Utilizzato come segnale di completamento o errore. Può essere trattato come una versione reattiva di Runnable.

• MayBe- Nessun elemento o 1 elemento emesso. Può essere trattato come una versione reattiva di Opzionale.

Di seguito sono riportati i metodi convenienti per creare osservabili nella classe Observable.

• just(T item) - Restituisce un osservabile che segnala l'elemento dato (riferimento costante) e quindi completa.

• fromIterable(Iterable source) - Converte una sequenza Iterable in un ObservableSource che emette gli elementi nella sequenza.

• fromArray(T... items) - Converte un array in un ObservableSource che emette gli elementi nell'array.

• fromCallable(Callable supplier) - Restituisce un Observable che, quando un osservatore si iscrive ad esso, richiama una funzione specificata e quindi emette il valore restituito da quella funzione.

• fromFuture(Future future) - Converte un futuro in una ObservableSource.

• interval(long initialDelay, long period, TimeUnit unit) - Restituisce un Observable che emette uno 0L dopo il InitialDelay e numeri sempre crescenti dopo ogni periodo di tempo successivo.

La classe Single rappresenta la risposta a valore singolo. Un singolo osservabile può emettere solo un singolo valore riuscito o un errore. Non emette l'evento onComplete.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.Single<T> classe -

public abstract class Single<T>
   extends Object
      implements SingleSource<T>

Protocollo

Di seguito è riportato il protocollo sequenziale utilizzato da Single Observable:

onSubscribe (onSuccess | onError)?

Unico esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {
      //Create the observable
      Single<String> testSingle = Single.just("Hello World");

      //Create an observer
      Disposable disposable = testSingle
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(
         new DisposableSingleObserver<String>() {

         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Hello World

La classe MayBe rappresenta la risposta differita. Può essere osservabile può emettere un singolo valore riuscito o nessun valore.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.Single<T> classe -

public abstract class Maybe<T>
   extends Object
      implements MaybeSource<T>

Protocollo

Di seguito è riportato il protocollo sequenziale utilizzato da MayBe Observable:

onSubscribe (onSuccess | onError | OnComplete)?

MayBe Example

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Maybe;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableMaybeObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      //Create an observer
      Disposable disposable = Maybe.just("Hello World")
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(new DisposableMaybeObserver<String>() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }

         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Hello World

La classe Completable rappresenta la risposta differita. L'osservabile completabile può indicare un completamento riuscito o un errore.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.Completable classe -

public abstract class Completable
extends Object
implements CompletableSource

Protocollo

Di seguito è riportato il protocollo sequenziale utilizzato da Completable Observable:

onSubscribe (onError | onComplete)?

Esempio completabile

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Completable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableCompletableObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {

      //Create an observer
      Disposable disposable = Completable.complete()
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(new DisposableCompletableObserver() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }
         @Override
         public void onStart() {
            System.out.println("Started!");
         }
         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Started!
Done!

La classe CompositeDisposable rappresenta un contenitore che può contenere più articoli usa e getta e offre una complessità O (1) di aggiunta e rimozione di articoli usa e getta.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.disposables.CompositeDisposable classe -

public final class CompositeDisposable
extends Object
implements Disposable, io.reactivex.internal.disposables.DisposableContainer

Esempio CompositeDisposable

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Maybe;
import io.reactivex.Single;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableMaybeObserver;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {    
      CompositeDisposable compositeDisposable = new CompositeDisposable();

      //Create an Single observer 
      Disposable disposableSingle = Single.just("Hello World")
      .delay(2, TimeUnit.SECONDS, Schedulers.io())
      .subscribeWith(
      new DisposableSingleObserver<String>() {
         @Override
         public void onError(Throwable e) {
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }
      }); 

      //Create an observer
      Disposable disposableMayBe = Maybe.just("Hi")
      .delay(2, TimeUnit.SECONDS, Schedulers.io())
      .subscribeWith(new DisposableMaybeObserver<String>() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }

         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 

      Thread.sleep(3000);

      compositeDisposable.add(disposableSingle);
      compositeDisposable.add(disposableMayBe);

      //start observing
      compositeDisposable.dispose();
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Hello World
Hi

Di seguito sono riportati gli operatori utilizzati per creare un osservabile.

Sr.No.	Operatore e descrizione
1	Create Crea un Observable da zero e consente al metodo dell'osservatore di chiamare a livello di codice.
2	Defer Non creare un osservabile fino a quando un osservatore non si abbona. Crea un nuovo osservabile per ogni osservatore.
3	Empty/Never/Throw Crea un osservabile con un comportamento limitato.
4	From Converte una struttura oggetto / dati in un osservabile.
5	Interval Crea un Osservabile che emette numeri interi in sequenza con un intervallo di intervallo di tempo specificato.
6	Just Converte una struttura oggetto / dati in un osservabile per emettere lo stesso o lo stesso tipo di oggetti.
7	Range Crea un Osservabile che emette numeri interi in una sequenza di un dato intervallo.
8	Repeat Crea un osservabile che emette numeri interi in sequenza ripetutamente.
9	Start Crea un oggetto Observable per emettere il valore restituito di una funzione.
10	Timer Crea un osservabile per emettere un singolo oggetto dopo un determinato ritardo.

Creazione di un esempio di operatore

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using fromArray operator to create an Observable
public class ObservableTester  {
   public static void main(String[] args) { 
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .map(String::toUpperCase)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

ABCDEFG

Di seguito sono riportati gli operatori che vengono utilizzati per trasformare un elemento emesso da un osservabile.

Sr.No.	Operatore e descrizione
1	Buffer Raccoglie periodicamente gli elementi da Observable in bundle e quindi emette i bundle anziché gli articoli.
2	FlatMap Utilizzato in osservabili annidati. Trasforma gli oggetti in osservabili. Quindi appiattisci gli elementi in un singolo osservabile.
3	GroupBy Dividi un osservabile in un insieme di osservabili organizzati per chiave per emettere diversi gruppi di elementi.
4	Map Applicare una funzione a ciascun elemento emesso per trasformarlo.
5	Scan Applicare una funzione a ciascun elemento emesso, in sequenza e quindi emettere il valore successivo.
6	Window Raccoglie periodicamente gli elementi da Observable a Observable e quindi emette le finestre anziché gli elementi.

Esempio di operatore di trasformazione

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using map operator to transform an Observable
public class ObservableTester  { 
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .map(String::toUpperCase)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

ABCDEFG

Di seguito sono riportati gli operatori che vengono utilizzati per emettere selettivamente elementi da un osservabile.

Sr.No.	Operatore e descrizione
1	Debounce Emette elementi solo quando si verifica il timeout senza emettere un altro elemento.
2	Distinct Emette solo oggetti unici.
3	ElementAt emette solo un elemento con indice n emesso da un Observable.
4	Filter Emette solo quegli elementi che passano la funzione predicato data.
5	First Emette il primo elemento o il primo elemento che ha superato i criteri specificati.
6	IgnoreElements Non emette alcun oggetto da Observable ma segna il completamento.
7	Last Emette l'ultimo elemento da Observable.
8	Sample Emette l'elemento più recente con un determinato intervallo di tempo.
9	Skip Salta i primi n elementi da un osservabile.
10	SkipLast Salta gli ultimi n elementi da un osservabile.
11	Take prende i primi n elementi da un osservabile.
12	TakeLast prende gli ultimi n elementi da un osservabile.

Esempio di operatore di filtro

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using take operator to filter an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .take(2)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

ab

Di seguito sono riportati gli operatori che vengono utilizzati per creare un singolo osservabile da più osservabili.

Sr.No.	Operatore e descrizione
1	And/Then/When Combina set di articoli utilizzando gli intermediari Pattern e Plan.
2	CombineLatest Combina l'ultimo elemento emesso da ciascun osservabile tramite una funzione specificata ed emetti l'elemento risultante.
3	Join Combina elementi emessi da due osservabili se emessi durante l'intervallo di tempo del secondo elemento emesso da osservabili.
4	Merge Combina gli elementi emessi di osservabili.
5	StartWith Emette una sequenza di elementi specificata prima di iniziare a emettere gli elementi dalla fonte Observable
6	Switch Emette gli elementi più recenti emessi da Observables.
7	Zip Combina elementi di osservabili in base alla funzione ed emette gli elementi risultanti.

Esempio di combinazione di operatori

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using combineLatest operator to combine Observables
public class ObservableTester {
   public static void main(String[] args) {    
      Integer[] numbers = { 1, 2, 3, 4, 5, 6};
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable1 = Observable.fromArray(letters);
      Observable<Integer> observable2 = Observable.fromArray(numbers);
      Observable.combineLatest(observable1, observable2, (a,b) -> a + b)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

g1g2g3g4g5g6

Di seguito sono riportati gli operatori che sono spesso utili con gli osservabili.

Sr.No.	Operatore e descrizione
1	Delay Registra l'azione per gestire gli eventi osservabili del ciclo di vita.
2	Materialize/Dematerialize Rappresenta l'elemento emesso e la notifica inviata.
3	ObserveOn Specificare lo scheduler da osservare.
4	Serialize Forza Observable per effettuare chiamate serializzate.
5	Subscribe Agisci sulle emissioni di oggetti e notifiche come se fosse completo da un osservabile
6	SubscribeOn Specificare lo scheduler che deve essere utilizzato da un Observable quando è sottoscritto.
7	TimeInterval Convertire un osservabile per emettere indicazioni del tempo trascorso tra le emissioni.
8	Timeout Emette una notifica di errore se il tempo specificato si verifica senza l'emissione di alcun elemento.
9	Timestamp Allega timestamp a ogni elemento emesso.
9	Using Crea una risorsa usa e getta o la stessa durata di quella di Observable.

Esempio di operatore di utilità

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using subscribe operator to subscribe to an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable.subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

abcdefg

Di seguito sono riportati gli operatori che valutano uno o più Osservabili o elementi emessi.

Sr.No.	Operatore e descrizione
1	All Valuta tutti gli elementi emessi per soddisfare determinati criteri.
2	Amb Emette tutti gli elementi dal primo osservabile solo dati più osservabili.
3	Contains Controlla se un osservabile emette un particolare elemento o meno.
4	DefaultIfEmpty Emette l'elemento predefinito se Observable non emette nulla.
5	SequenceEqual Controlla se due osservabili emettono la stessa sequenza di elementi.
6	SkipUntil Scarta gli oggetti emessi dal primo Observable fino a quando un secondo Observable non emette un oggetto.
7	SkipWhile Scarta gli elementi emessi da un Observable fino a quando una data condizione non diventa falsa.
8	TakeUntil Scarta gli elementi emessi da un Observable dopo che un secondo Observable emette un elemento o termina.
9	TakeWhile Scarta gli elementi emessi da un Observable dopo che una condizione specificata diventa falsa.

Esempio di operatore condizionale

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using defaultIfEmpty operator to operate on an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      final StringBuilder result = new StringBuilder();
      Observable.empty()
      .defaultIfEmpty("No Data")
      .subscribe(s -> result.append(s));
      System.out.println(result);
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result1 = new StringBuilder();
      Observable.fromArray(letters)
      .firstElement()
      .defaultIfEmpty("No data")   
      .subscribe(s -> result1.append(s));
      System.out.println(result1);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

No Data
a

Di seguito sono riportati gli operatori che operano su interi elementi emessi da un Osservabile.

Sr.No.	Operatore e descrizione
1	Average Valuta le medie di tutti gli elementi ed emette il risultato.
2	Concat Emette tutti gli elementi da più osservabili senza interleaving.
3	Count Conta tutti gli elementi ed emette il risultato.
4	Max Valuta l'elemento con valore massimo di tutti gli articoli ed emette il risultato.
5	Min Valuta l'elemento con valore minimo di tutti gli articoli ed emette il risultato.
6	Reduce Applica una funzione su ogni elemento e restituisci il risultato.
7	Sum Valuta la somma di tutti gli elementi ed emette il risultato.

Esempio di operatore matematico

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using concat operator to operate on multiple Observables
public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {    
      Integer[] numbers = { 1, 2, 3, 4, 5, 6};
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable1 = Observable.fromArray(letters);
      Observable<Integer> observable2 = Observable.fromArray(numbers);
      Observable.concat(observable1, observable2)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

abcdefg123456

Di seguito sono riportati gli operatori che hanno un controllo più preciso sull'abbonamento.

Sr.No.	Operatore e descrizione
1	Connect Indica a un osservabile collegabile di emettere elementi ai suoi abbonati.
2	Publish Converte un osservabile in osservabile collegabile.
3	RefCount Converte un osservabile collegabile in un osservabile ordinario.
4	Replay Assicurati che la stessa sequenza di elementi emessi sia visualizzata da ciascun abbonato, anche dopo che l'Osservabile ha iniziato a emettere elementi e gli abbonati si iscrivono in un secondo momento.

Esempio di operatore collegabile

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.observables.ConnectableObservable;
//Using connect operator on a ConnectableObservable
public class ObservableTester {
   public static void main(String[] args) {
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      ConnectableObservable<String> connectable = Observable.fromArray(letters).publish();      
      connectable.subscribe(letter -> result.append(letter));
      System.out.println(result.length());
      connectable.connect();
      System.out.println(result.length());
      System.out.println(result);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

0
7
abcdefg

Secondo il Reactive, un Soggetto può agire sia come Osservabile che come Osservatore.

Un Soggetto è una sorta di bridge o proxy disponibile in alcune implementazioni di ReactiveX che agisce sia come osservatore che come osservabile. Poiché è un osservatore, può iscriversi a uno o più osservabili e, poiché è un osservabile, può passare attraverso gli elementi che osserva riemettendoli e può anche emettere nuovi elementi.

Esistono quattro tipi di soggetti:

Sr.No.	Oggetto e descrizione
1	Publish Subject Emette solo quegli elementi che vengono emessi dopo il momento dell'abbonamento.
2	Replay Subject Emette tutti gli elementi emessi dalla sorgente Observable indipendentemente da quando ha sottoscritto Observable.
3	Behavior Subject Al momento dell'abbonamento, emette l'elemento più recente, quindi continua a emettere l'elemento emesso dalla fonte Observable.
4	Async Subject Emette l'ultimo elemento emesso dalla sorgente Observable dopo che ha completato l'emissione.

PublishSubject emette elementi agli osservatori attualmente iscritti ed eventi terminali agli osservatori attuali o in ritardo.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.subjects.PublishSubject<T> classe -

public final class PublishSubject<T>
extends Subject<T>

PublishSubject Example

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.PublishSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      PublishSubject<String> subject = PublishSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be abcd 
      System.out.println(result1);
      //Output will be d only
      //as subscribed after c item emitted.
      System.out.println(result2);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

abcd
d

BehaviorSubject emette l'elemento più recente che ha osservato e quindi tutti gli elementi osservati successivi a ciascun Observer sottoscritto.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.subjects.BehaviorSubject<T> classe -

public final class BehaviorSubject<T>
extends Subject<T>

BehaviorSubject Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.BehaviorSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         
      BehaviorSubject<String> subject =  BehaviorSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();
      //Output will be abcd
      System.out.println(result1);
      //Output will be cd being BehaviorSubject 
      //(c is last item emitted before subscribe)
      System.out.println(result2);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

abcd
cd

ReplaySubject riproduce eventi / oggetti agli osservatori attuali e in ritardo.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.subjects.ReplaySubject<T> classe -

public final class ReplaySubject<T>
extends Subject<T>

Esempio di ReplaySubject

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.ReplaySubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      ReplaySubject<String> subject = ReplaySubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be abcd
      System.out.println(result1);
      //Output will be abcd being ReplaySubject
      //as ReplaySubject emits all the items
      System.out.println(result2);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

abcd
abcd

AsyncSubject emette l'unico ultimo valore seguito da un evento di completamento o l'errore ricevuto a Observers.

Dichiarazione di classe

Di seguito è riportata la dichiarazione per io.reactivex.subjects.AsyncSubject<T> classe -

public final class  AsyncSubject<T>
extends Subject<T>

AsyncSubject Example

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects. AsyncSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      AsyncSubject<String> subject =  AsyncSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be d being the last item emitted
      System.out.println(result1);
      //Output will be d being the last item emitted     
      System.out.println(result2);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

d
d

Gli scheduler vengono utilizzati in ambienti multi-threading per lavorare con operatori Observable.

Secondo il Reactive, Scheduler vengono utilizzati per pianificare il modo in cui la catena di operatori verrà applicata a thread diversi.

Per impostazione predefinita, un Observable e la catena di operatori a cui applichi faranno il suo lavoro, e notificheranno i suoi osservatori, sullo stesso thread su cui viene chiamato il suo metodo di iscrizione. L'operatore SubscribeOn modifica questo comportamento specificando un diverso Scheduler su cui deve operare Observable. L'operatore ObserveOn specifica un diverso Scheduler che Observable utilizzerà per inviare notifiche ai suoi osservatori.

Ci sono i seguenti tipi di Scheduler disponibili in RxJava:

Sr.No.	Scheduler e descrizione
1	Schedulers.computation() Crea e restituisce uno Scheduler destinato al lavoro computazionale. Il conteggio dei thread da pianificare dipende dalle CPU presenti nel sistema. È consentito un thread per CPU. Ideale per loop di eventi o operazioni di callback.
2	Schedulers.io() Crea e restituisce uno Scheduler destinato al lavoro associato a IO. Il pool di thread può estendersi secondo necessità.
3	Schedulers.newThread() Crea e restituisce uno Scheduler che crea un nuovo Thread per ogni unità di lavoro.
4	Schedulers.trampoline() Crea e restituisce uno Scheduler che accoda il lavoro sul thread corrente da eseguire dopo il completamento del lavoro corrente.
4	Schedulers.from(java.util.concurrent.Executor executor) Converte un Executor in una nuova istanza dello Scheduler.

Il metodo Schedulers.trampoline () crea e restituisce uno Scheduler che esegue le code sul thread corrente da eseguire dopo il completamento del lavoro corrente.

Schedulers.trampoline () Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.trampoline()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Processing Thread main
Receiver Thread main, Item length 1
Processing Thread main
Receiver Thread main, Item length 2
Processing Thread main
Receiver Thread main, Item length 3

Il metodo Schedulers.newThread () crea e restituisce uno Scheduler che crea un nuovo Thread per ogni unità di lavoro.

Schedulers.newThread () Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.newThread()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Processing Thread RxNewThreadScheduler-1
Receiver Thread RxNewThreadScheduler-1, Item length 1
Processing Thread RxNewThreadScheduler-2
Receiver Thread RxNewThreadScheduler-2, Item length 2
Processing Thread RxNewThreadScheduler-3
Receiver Thread RxNewThreadScheduler-3, Item length 3

Il metodo Schedulers.computation () crea e restituisce uno Scheduler destinato al lavoro di calcolo. Il conteggio dei thread da pianificare dipende dalle CPU presenti nel sistema. È consentito un thread per CPU. Ideale per loop di eventi o operazioni di callback.

Schedulers.computation () Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.computation()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Processing Thread RxComputationThreadPool-1
Receiver Thread RxComputationThreadPool-1, Item length 1
Processing Thread RxComputationThreadPool-2
Receiver Thread RxComputationThreadPool-2, Item length 2
Processing Thread RxComputationThreadPool-3
Receiver Thread RxComputationThreadPool-3, Item length 3

Il metodo Schedulers.io () crea e restituisce uno Scheduler destinato al lavoro associato a IO. Il pool di thread può estendersi secondo necessità. Ideale per operazioni intensive di I / O.

Schedulers.io () Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.io()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 1
Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 2
Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 3

Il metodo Schedulers.from (Executor) converte un Executor in una nuova istanza dello Scheduler.

Schedulers.from (Executor) Esempio

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import java.util.Random;
import java.util.concurrent.Executors;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.from(Executors.newFixedThreadPool(3))))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Processing Thread pool-1-thread-1
Processing Thread pool-3-thread-1
Receiver Thread pool-1-thread-1, Item length 1
Processing Thread pool-4-thread-1
Receiver Thread pool-4-thread-1, Item length 3
Receiver Thread pool-3-thread-1, Item length 2

L'operatore di buffering consente di raccogliere gli elementi emessi da un Observable in un elenco o bundle ed emettere quei bundle invece di elementi. Nell'esempio seguente, abbiamo creato un Observable per emettere 9 elementi e utilizzando il buffering, 3 elementi verranno emessi insieme.

Esempio di buffering

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      Observable<Integer> observable = Observable.just(1, 2, 3, 4,
         5, 6, 7, 8, 9);

      observable.subscribeOn(Schedulers.io())
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .buffer(3)
         .subscribe(new Observer<List<Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
               System.out.println("Subscribed");
            }
            @Override
            public void onNext(List<Integer> integers) {
               System.out.println("onNext: ");
               for (Integer value : integers) {
                  System.out.println(value);
               }
            }
            @Override
            public void onError(Throwable e) {
               System.out.println("Error");
            }

            @Override
            public void onComplete() {
               System.out.println("Done! ");
            }
         });
      Thread.sleep(3000);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Subscribed
onNext: 
1
2
3
onNext: 
4
5
6
onNext: 
7
8
9
Done!

L'operatore di windowing funziona in modo simile all'operatore di buffer ma consente di raccogliere gli elementi emessi da un Observable in un altro osservabile invece della raccolta ed emettere quegli osservabili invece delle raccolte. Nell'esempio seguente, abbiamo creato un Observable per emettere 9 elementi e utilizzando l'operatore di finestra, 3 Observable verranno emessi insieme.

Esempio di finestre

Crea il seguente programma Java utilizzando qualsiasi editor di tua scelta, ad esempio C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      Observable<Integer> observable = Observable.just(1, 2, 3, 4,
         5, 6, 7, 8, 9);

      observable.subscribeOn(Schedulers.io())
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .window(3)
         .subscribe(new Observer<Observable<Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
               System.out.println("Subscribed");
            }
            @Override
            public void onNext(Observable<Integer> integers) {
               System.out.println("onNext: ");
               integers.subscribe(value -> System.out.println(value));
            }
            @Override
            public void onError(Throwable e) {
               System.out.println("Error");
            }

            @Override
            public void onComplete() {
               System.out.println("Done! ");
            }
         });
      Thread.sleep(3000);
   }
}

Verifica il risultato

Compila la classe usando javac compilatore come segue -

C:\RxJava>javac ObservableTester.java

Ora esegui ObservableTester come segue:

C:\RxJava>java ObservableTester

Dovrebbe produrre il seguente output:

Subscribed
onNext: 
1
2
3
onNext: 
4
5
6
onNext: 
7
8
9
Done!