RxJava - Guia rápido

RxJava é uma extensão baseada em Java do ReactiveX. Ele fornece implementação ou projeto ReactiveX em Java. A seguir estão as principais características do RxJava.

• Estende o padrão do observador.

• Suporta sequências de dados / eventos.

• Fornece operadores para compor sequências de forma declarativa.

• Lida com threading, sincronização, thread-safety e estruturas de dados concorrentes internamente.

O que é ReactiveX?

ReactiveX é um projeto que visa fornecer conceito de programação reativa para várias linguagens de programação. A programação reativa se refere ao cenário em que o programa reage à medida que os dados aparecem. É um conceito de programação baseado em eventos e os eventos podem se propagar para observadores de registradores.

Conforme Reactive, eles combinaram o melhor do padrão Observer, do padrão Iterator e do padrão funcional.

O padrão Observer bem feito. ReactiveX é uma combinação das melhores ideias do padrão Observer, do padrão Iterator e da programação funcional.

Programação Funcional

A programação funcional gira em torno da construção do software usando funções puras. Uma função pura não depende do estado anterior e sempre retorna o mesmo resultado para os mesmos parâmetros passados. As funções puras ajudam a evitar problemas associados a objetos compartilhados, dados mutáveis ​​e efeitos colaterais geralmente predominantes em ambientes de multi-threading.

Programação reativa

A programação reativa se refere à programação orientada a eventos, em que os fluxos de dados chegam de forma assíncrona e são processados ​​quando chegam.

Programação Reativa Funcional

RxJava implementa ambos os conceitos juntos, onde os dados dos fluxos mudam ao longo do tempo e a função do consumidor reage de acordo.

O Manifesto Reativo

Manifesto reativo é um documento online que afirma o alto padrão dos sistemas de software de aplicação. De acordo com o manifesto, a seguir estão os principais atributos de um software reativo -

• Responsive - Deve sempre responder em tempo hábil.

• Message Driven - Deve usar a passagem de mensagens assíncrona entre os componentes para que eles mantenham um acoplamento fraco.

• Elastic - Deve permanecer ágil mesmo sob carga elevada.

• Resilient - Deve permanecer responsivo mesmo se algum componente falhar.

Principais componentes do RxJava

RxJava tem dois componentes principais: Observables e Observer.

• Observable - Representa um objeto semelhante ao Stream que pode emitir zero ou mais dados, pode enviar mensagem de erro, cuja velocidade pode ser controlada durante a emissão de um conjunto de dados, pode enviar dados finitos como infinitos.

• Observer- Assina os dados de sequência do Observable e reage por item dos observáveis. Os observadores são notificados sempre que Observable emite dados. Um Observer trata os dados um por um.

Um observador nunca é notificado se os itens não estiverem presentes ou um retorno de chamada não for retornado para um item anterior.

Configuração de ambiente local

RxJava é uma biblioteca para Java, então o primeiro requisito é ter o JDK instalado em sua máquina.

Requisitos do sistema

JDK	1,5 ou superior.
Memória	Sem requisitos mínimos.
Espaço em disco	Sem requisitos mínimos.
Sistema operacional	Sem requisitos mínimos.

Etapa 1 - Verifique a instalação do Java em sua máquina

Em primeiro lugar, abra o console e execute um comando java baseado no sistema operacional em que está trabalhando.

SO	Tarefa	Comando
janelas	Abrir console de comando	c: \> versão java
Linux	Abra o terminal de comando	$ java -version
Mac	Terminal aberto	máquina: <joseph $ java -version

Vamos verificar a saída de todos os sistemas operacionais -

SO	Resultado
janelas	versão java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)
Linux	versão java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)
Mac	versão java "1.8.0_101" Java (TM) SE Runtime Environment (build 1.8.0_101)

Se você não tiver o Java instalado em seu sistema, baixe o Java Software Development Kit (SDK) do seguinte link https://www.oracle.com. Estamos assumindo o Java 1.8.0_101 como a versão instalada para este tutorial.

Etapa 2 - Definir ambiente JAVA

Colocou o JAVA_HOMEvariável de ambiente para apontar para o local do diretório base onde o Java está instalado em sua máquina. Por exemplo.

SO	Resultado
janelas	Defina a variável de ambiente JAVA_HOME para C: \ Program Files \ Java \ jdk1.8.0_101
Linux	export JAVA_HOME = / usr / local / java-current
Mac	export JAVA_HOME = / Library / Java / Home

Anexe a localização do compilador Java ao caminho do sistema.

SO	Resultado
janelas	Anexar a string C:\Program Files\Java\jdk1.8.0_101\bin no final da variável do sistema, Path.
Linux	exportar PATH = $ PATH: $ JAVA_HOME / bin /
Mac	não requerido

Verifique a instalação do Java usando o comando java -version como explicado acima.

Etapa 3 - Baixe o arquivo RxJava2

Baixe a versão mais recente do arquivo jar RxJava de RxJava @ MVNRepository e sua dependência Reactive Streams @ MVNRepository . No momento de escrever este tutorial, baixamos rxjava-2.2.4.jar, reactive-streams-1.0.2.jar e os copiamos para a pasta C: \> RxJava.

SO	Nome do arquivo
janelas	rxjava-2.2.4.jar, reativo-streams-1.0.2.jar
Linux	rxjava-2.2.4.jar, reativo-streams-1.0.2.jar
Mac	rxjava-2.2.4.jar, reativo-streams-1.0.2.jar

Etapa 4 - Definir o ambiente RxJava

Colocou o RX_JAVAvariável de ambiente para apontar para o local do diretório base onde o jar RxJava está armazenado em sua máquina. Vamos supor que armazenamos rxjava-2.2.4.jar e reactive-streams-1.0.2.jar na pasta RxJava.

Sr. Não	SO e descrição
1	Windows Defina a variável de ambiente RX_JAVA para C: \ RxJava
2	Linux exportar RX_JAVA = / usr / local / RxJava
3	Mac exportar RX_JAVA = / Library / RxJava

Etapa 5 - Definir Variável CLASSPATH

Colocou o CLASSPATH variável de ambiente para apontar para o local do jar RxJava.

Sr. Não	SO e descrição
1	Windows Defina a variável de ambiente CLASSPATH como% CLASSPATH%;% RX_JAVA% \ rxjava-2.2.4.jar;% RX_JAVA% \ reactive-streams-1.0.2.jar;.;
2	Linux export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: reactive-streams-1.0.2.jar :.
3	Mac export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: reactive-streams-1.0.2.jar :.

Etapa 6 - Teste a configuração do RxJava

Crie uma classe TestRx.java conforme mostrado abaixo -

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

Etapa 7 - Verifique o resultado

Compile as classes usando javac compilador da seguinte forma -

C:\RxJava>javac Tester.java

Verifique a saída.

Hello World!

Observables representa as fontes de dados onde Observers (Subscribers)escute-os. Resumindo, um Observável emite itens e um Assinante então consome esses itens.

Observável

• Observável fornece dados assim que o assinante começa a ouvir.

• Observável pode emitir qualquer número de itens.

• Observável pode emitir apenas um sinal de conclusão e nenhum item.

• Observável pode terminar com sucesso.

• Observável pode nunca terminar. por exemplo, um botão pode ser clicado qualquer número de vezes.

• Observável pode gerar erro a qualquer momento.

Assinante

• Observável pode ter vários assinantes.

• Quando um Observable emite um item, cada método onNext () do assinante é chamado.

• Quando um Observable termina de emitir itens, cada método onComplete () do assinante é chamado.

• Se um Observable emitir um erro, cada método onError () do assinante será chamado.

A seguir estão as classes básicas para criar observáveis.

• Flowable- 0..N flui, emite 0 ou n itens. Suporta fluxos reativos e contrapressão.

• Observable - 0..N flui, mas sem contrapressão.

• Single- 1 item ou erro. Pode ser tratado como uma versão reativa da chamada de método.

• Completable- Nenhum item emitido. Usado como um sinal de conclusão ou erro. Pode ser tratado como uma versão reativa do Runnable.

• MayBe- Nenhum item ou 1 item emitido. Pode ser tratada como uma versão reativa de Opcional.

A seguir estão os métodos convenientes para criar observáveis ​​na classe Observable.

• just(T item) - Retorna um Observable que sinaliza o item fornecido (referência constante) e então é concluído.

• fromIterable(Iterable source) - Converte uma sequência Iterable em um ObservableSource que emite os itens na sequência.

• fromArray(T... items) - Converte um Array em um ObservableSource que emite os itens do Array.

• fromCallable(Callable supplier) - Retorna um Observable que, quando um observador se inscreve nele, invoca uma função que você especifica e emite o valor retornado dessa função.

• fromFuture(Future future) - Converte um futuro em uma fonte observável.

• interval(long initialDelay, long period, TimeUnit unit) - Retorna um Observable que emite um 0L após o initialDelay e números sempre crescentes após cada período de tempo posterior.

A classe Single representa a resposta de valor único. Observável único pode emitir apenas um único valor bem-sucedido ou um erro. Ele não emite o evento onComplete.

Declaração de Classe

A seguir está a declaração para io.reactivex.Single<T> classe -

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

Protocolo

A seguir está o protocolo sequencial que opera o único observável -

onSubscribe (onSuccess | onError)?

Exemplo Único

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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();
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

Hello World

A classe MayBe representa uma resposta adiada. MayBe observable pode emitir um único valor de sucesso ou nenhum valor.

Declaração de Classe

A seguir está a declaração para io.reactivex.Single<T> classe -

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

Protocolo

A seguir está o protocolo sequencial que MayBe Observable opera -

onSubscribe (onSuccess | onError | OnComplete)?

MayBe Example

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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();
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

Hello World

A classe Completable representa uma resposta adiada. O observável completável pode indicar uma conclusão bem-sucedida ou um erro.

Declaração de Classe

A seguir está a declaração para io.reactivex.Completable classe -

public abstract class Completable
extends Object
implements CompletableSource

Protocolo

A seguir está o protocolo sequencial que o Completable Observable opera -

onSubscribe (onError | onComplete)?

Exemplo Completável

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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();
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

Started!
Done!

A classe CompositeDisposable representa um contêiner que pode conter vários descartáveis ​​e oferece complexidade O (1) de adição e remoção de descartáveis.

Declaração de Classe

A seguir está a declaração para io.reactivex.disposables.CompositeDisposable classe -

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

Exemplo de CompositeDisposable

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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();
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

Hello World
Hi

A seguir estão os operadores que são usados ​​para criar um Observable.

Sr. Não.	Operador e descrição
1	Create Cria um Observable do zero e permite que o método do observador chame programaticamente.
2	Defer Não crie um Observable até que um observador se inscreva. Cria um novo observável para cada observador.
3	Empty/Never/Throw Cria um observável com comportamento limitado.
4	From Converte uma estrutura de objeto / dados em um Observable.
5	Interval Cria um número inteiro de emissão observável em sequência com um intervalo de intervalo de tempo especificado.
6	Just Converte uma estrutura de objeto / dados em um Observable para emitir o mesmo ou o mesmo tipo de objetos.
7	Range Cria um número inteiro de emissão observável na sequência de um determinado intervalo.
8	Repeat Cria um observável emitindo inteiros em sequência repetidamente.
9	Start Cria um Observable para emitir o valor de retorno de uma função.
10	Timer Cria um Observable para emitir um único item após um determinado atraso.

Criando exemplo de operador

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

ABCDEFG

A seguir estão os operadores que são usados ​​para transformar um item emitido de um Observable.

Sr. Não.	Operador e descrição
1	Buffer Reúne itens do Observável em pacotes periodicamente e, em seguida, emite os pacotes em vez de itens.
2	FlatMap Usado em observáveis ​​aninhados. Transforma itens em observáveis. Em seguida, achatar os itens em um único Observável.
3	GroupBy Divida um Observável em um conjunto de Observáveis ​​organizado por chave para emitir diferentes grupos de itens.
4	Map Aplique uma função a cada item emitido para transformá-lo.
5	Scan Aplique uma função a cada item emitido, sequencialmente e, a seguir, emita o valor sucessivo.
6	Window Reúne itens de janelas Observáveis ​​em janelas Observáveis ​​periodicamente e, a seguir, emite as janelas em vez de itens.

Exemplo de operador de transformação

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

ABCDEFG

A seguir estão os operadores que são usados ​​para emitir seletivamente item (ns) de um Observável.

Sr. Não.	Operador e descrição
1	Debounce Emite itens apenas quando o tempo limite ocorre sem emitir outro item.
2	Distinct Emite apenas itens exclusivos.
3	ElementAt emite apenas o item no índice n emitido por um Observable.
4	Filter Emite apenas os itens que passam pela função de predicado fornecida.
5	First Emite o primeiro item ou primeiro item que passou nos critérios fornecidos.
6	IgnoreElements Não emite nenhum item do Observável, mas marca a conclusão.
7	Last Emite o último elemento do Observable.
8	Sample Emite o item mais recente com determinado intervalo de tempo.
9	Skip Ignora os primeiros n itens de um Observável.
10	SkipLast Pula os últimos n itens de um Observável.
11	Take pega os primeiros n itens de um Observable.
12	TakeLast pega os últimos n itens de um Observable.

Exemplo de operador de filtragem

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

ab

A seguir estão os operadores que são usados ​​para criar um único observável a partir de vários observáveis.

Sr. Não.	Operador e descrição
1	And/Then/When Combine conjuntos de itens usando intermediários de padrão e plano.
2	CombineLatest Combine o último item emitido por cada Observable por meio de uma função especificada e emita o item resultante.
3	Join Combine itens emitidos por dois Observáveis ​​se emitidos durante o período do segundo item Observável emitido.
4	Merge Combina os itens emitidos de Observáveis.
5	StartWith Emita uma sequência especificada de itens antes de começar a emitir os itens da fonte Observável
6	Switch Emite os itens mais recentes emitidos por Observables.
7	Zip Combina itens de Observáveis ​​com base na função e emite os itens resultantes.

Exemplo de operador de combinação

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

g1g2g3g4g5g6

A seguir estão os operadores que geralmente são úteis com Observáveis.

Sr. Não.	Operador e descrição
1	Delay Registre a ação para lidar com eventos de ciclo de vida observáveis.
2	Materialize/Dematerialize Representa o item emitido e a notificação enviada.
3	ObserveOn Especifique o planejador a ser observado.
4	Serialize Força Observable a fazer chamadas serializadas.
5	Subscribe Operar sobre as emissões de itens e notificações como concluídas de um Observável
6	SubscribeOn Especifique o planejador a ser usado por um Observable quando ele for inscrito.
7	TimeInterval Converta um Observável para emitir indicações da quantidade de tempo decorrido entre as emissões.
8	Timeout Emite notificação de erro se o tempo especificado ocorrer sem a emissão de nenhum item.
9	Timestamp Anexe o carimbo de data / hora a cada item emitido.
9	Using Cria um recurso descartável ou mesmo tempo de vida do Observable.

Exemplo de operador de serviço público

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

abcdefg

A seguir estão os operadores que avaliam um ou vários Observáveis ​​ou itens emitidos.

Sr. Não.	Operador e descrição
1	All Avalia todos os itens emitidos para atender a determinados critérios.
2	Amb Emite todos os itens do primeiro Observable apenas dados vários Observables.
3	Contains Verifica se um Observable emite um item específico ou não.
4	DefaultIfEmpty Emite o item padrão se Observable não emitir nada.
5	SequenceEqual Verifica se dois Observáveis ​​emitem a mesma sequência de itens.
6	SkipUntil Descarta os itens emitidos pelo primeiro Observável até que um segundo Observável emita um item.
7	SkipWhile Descarte itens emitidos por um Observável até que uma determinada condição se torne falsa.
8	TakeUntil Descarta os itens emitidos por um Observable depois que um segundo Observable emite um item ou termina.
9	TakeWhile Descarte os itens emitidos por um Observável após uma condição especificada se tornar falsa.

Exemplo de operador condicional

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

No Data
a

A seguir estão os operadores que operam em itens inteiros emitidos por um Observable.

Sr. Não.	Operador e descrição
1	Average Avalia as médias de todos os itens e emite o resultado.
2	Concat Emite todos os itens de vários Observáveis ​​sem intercalação.
3	Count Conta todos os itens e emite o resultado.
4	Max Avalia o item com valor máximo de todos os itens e emite o resultado.
5	Min Avalia o item de valor mínimo de todos os itens e emite o resultado.
6	Reduce Aplique uma função em cada item e retorne o resultado.
7	Sum Avalia soma de todos os itens e emite o resultado.

Exemplo de operador matemático

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

abcdefg123456

A seguir estão as operadoras que têm controle mais preciso sobre a assinatura.

Sr. Não.	Operador e descrição
1	Connect Instrua um Observable conectável a emitir itens para seus assinantes.
2	Publish Converte um Observável em um Observável conectável.
3	RefCount Converte um Observável conectável em um Observável comum.
4	Replay Garanta que a mesma sequência de itens emitidos seja vista por cada assinante, mesmo depois que o Observable começar a emitir itens e os assinantes se inscreverem posteriormente.

Exemplo de operador conectável

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

0
7
abcdefg

Conforme Reactive, um Assunto pode atuar como Observável e Observador.

Um Assunto é uma espécie de ponte ou proxy que está disponível em algumas implementações do ReactiveX que atua tanto como um observador quanto como um Observável. Por ser um observador, pode inscrever-se em um ou mais Observáveis ​​e, por ser um Observável, pode passar pelos itens que observa reemitindo-os e também pode emitir novos itens.

Existem quatro tipos de assuntos -

Sr. Não.	Assunto e descrição
1	Publish Subject Emite apenas os itens que são emitidos após o tempo de assinatura.
2	Replay Subject Emite todos os itens emitidos pela fonte Observable, independentemente de quando ela assinou o Observable.
3	Behavior Subject Após a assinatura, emite o item mais recente e, em seguida, continua a emitir o item emitido pela fonte Observável.
4	Async Subject Emite o último item emitido pela fonte Observável após completar a emissão.

PublishSubject emite itens para Observadores atualmente inscritos e eventos de terminal para Observadores atuais ou atrasados.

Declaração de Classe

A seguir está a declaração para io.reactivex.subjects.PublishSubject<T> classe -

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

Exemplo de PublishSubject

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

abcd
d

BehaviorSubject emite o item mais recente que observou e, em seguida, todos os itens subsequentes observados para cada Observador inscrito.

Declaração de Classe

A seguir está a declaração para io.reactivex.subjects.BehaviorSubject<T> classe -

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

BehaviorSubject Example

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

abcd
cd

ReplaySubject reproduz eventos / itens para Observadores atuais e atrasados.

Declaração de Classe

A seguir está a declaração para io.reactivex.subjects.ReplaySubject<T> classe -

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

Exemplo de ReplaySubject

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

abcd
abcd

AsyncSubject emite o único último valor seguido por um evento de conclusão ou o erro recebido para observadores.

Declaração de Classe

A seguir está a declaração para io.reactivex.subjects.AsyncSubject<T> classe -

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

Exemplo de AsyncSubject

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

d
d

Schedulers são usados ​​em ambiente multi-threading para trabalhar com operadores Observable.

Conforme Reactive, Scheduler são usados ​​para agendar como a cadeia de operadores se aplicará a diferentes threads.

Por padrão, um Observable e a cadeia de operadores que você aplica a ele farão seu trabalho e notificarão seus observadores no mesmo thread em que seu método Subscribe é chamado. O operador SubscribeOn altera esse comportamento especificando um Scheduler diferente no qual o Observable deve operar. O operador ObserveOn especifica um Scheduler diferente que o Observable usará para enviar notificações a seus observadores.

Existem seguintes tipos de Schedulers disponíveis no RxJava -

Sr. Não.	Programador e descrição
1	Schedulers.computation() Cria e retorna um Scheduler destinado ao trabalho computacional. A contagem de threads a serem agendados depende das CPUs presentes no sistema. Um thread é permitido por CPU. Melhor para loops de eventos ou operações de retorno de chamada.
2	Schedulers.io() Cria e retorna um Scheduler destinado ao trabalho vinculado ao IO. O pool de threads pode ser estendido conforme necessário.
3	Schedulers.newThread() Cria e retorna um Scheduler que cria um novo Thread para cada unidade de trabalho.
4	Schedulers.trampoline() Cria e retorna um Scheduler que enfileira o trabalho no thread atual para ser executado após a conclusão do trabalho atual.
4	Schedulers.from(java.util.concurrent.Executor executor) Converte um Executor em uma nova instância do Scheduler.

O método Schedulers.trampoline () cria e retorna um Scheduler que enfileira o trabalho no thread atual para ser executado após a conclusão do trabalho atual.

Schedulers.trampoline () Exemplo

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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;
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O método Schedulers.newThread () cria e retorna um Scheduler que cria um novo Thread para cada unidade de trabalho.

Schedulers.newThread () Exemplo

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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;
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O método Schedulers.computation () cria e retorna um Scheduler destinado ao trabalho computacional. A contagem de threads a serem agendados depende das CPUs presentes no sistema. Um thread é permitido por CPU. Melhor para loops de eventos ou operações de retorno de chamada.

Schedulers.computation () Exemplo

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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;
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O método Schedulers.io () cria e retorna um Scheduler destinado ao trabalho vinculado a E / S. O pool de threads pode ser estendido conforme necessário. Melhor para operações intensivas de E / S.

Schedulers.io () Exemplo

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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;
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O método Schedulers.from (Executor) converte um Executor em uma nova instância do Scheduler.

Schedulers.from (Executor) Exemplo

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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;
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O operador de buffer permite reunir itens emitidos por um Observable em uma lista ou pacotes e emitir esses pacotes em vez de itens. No exemplo abaixo, criamos um Observable para emitir 9 itens e usando buffering, 3 itens serão emitidos juntos.

Exemplo de Buffering

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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

O operador de janela funciona de forma semelhante ao operador de buffer, mas permite reunir itens emitidos por um Observable em outro observável em vez de coleção e emitir esses Observáveis ​​em vez de coleções. No exemplo abaixo, criamos um Observable para emitir 9 itens e, usando o operador de janela, 3 Observable serão emitidos juntos.

Exemplo de janela

Crie o seguinte programa Java usando qualquer editor de sua escolha em, digamos, 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);
   }
}

Verifique o resultado

Compile a classe usando javac compilador da seguinte forma -

C:\RxJava>javac ObservableTester.java

Agora execute o ObservableTester da seguinte maneira -

C:\RxJava>java ObservableTester

Ele deve produzir a seguinte saída -

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