2. Um exemplo de introdução

Meus primeiros contatos com o RxJava ocorreram por meio de cursos e tutoriais encontrados na internet. Além de a teoria utilizar conceitos aos quais eu não estava acostumado e que eu tinha dificuldade em compreender, eu, acima de tudo, não conseguia imaginar para que isso poderia servir na vida real. Vamos, portanto, começar apresentando um exemplo (simples, espero) em que o uso do RxJava traz uma simplificação real na escrita do código e, a partir daí, tentaremos identificar os elementos importantes dessa biblioteca.

A biblioteca RxJava baseia-se no seguinte conceito: um fluxo de elementos do tipo T Observable<T> é observado por um ou mais assinantes (assinantes, observadores, consumidores) Subscriber<T>. A biblioteca RxJava permite que o fluxo Observable<T> seja executado em um thread T1 e seu observador Subscriber<T> em um thread T2, sem que o desenvolvedorprecise se preocupar em gerenciar o ciclo de vida dessas threads e com problemas naturalmente complexos, como o compartilhamento de dados entre threads e a sincronização entre elas para executar uma tarefa global. Assim, ela facilita a programação assíncrona.

Um fluxo Observable<T> produz elementos do tipo T, observáveis à medida que são produzidos. Se o observador e o observável (termo que, por conveniência, designa o tipo Observable<T>) estiverem na mesma thread, então o observável só poderá produzir o elemento (i+1) quando o observador tiver consumido o elemento i. Há poucos casos em que essa arquitetura seja interessante. Se o observador e o observável não estiverem no mesmo thread, então o observável e seu observador têm comportamentos autônomos: o observável produz em seu próprio ritmo e o observador consome em seu próprio ritmo. É aí que reside o interesse da biblioteca. Até agora, sempre falamos de um único observador. Na realidade, um observável pode ter qualquer número de observadores.

2.1. A arquitetura do aplicativo de exemplo

A aplicação de exemplo possui a seguinte arquitetura:

• em [1], uma camada de serviço fornece listas de números aleatórios. Essa camada é executada no mesmo thread que o método [swing] que a utiliza. Ela, então, fornece seus números de forma síncrona;
• em [2], uma fina camada de adaptação implementada com RxJava permite apresentar à camada [swing] uma implementação assíncrona do mesmo serviço: esta pode ser executada em uma thread diferente daquela do método [swing] que a utiliza;
• a chamada [4] é síncrona, enquanto a chamada [5-6] é assíncrona;

O que queremos mostrar aqui é que a biblioteca Rx permite transformar facilmente uma interface síncrona em uma interface assíncrona. Por que isso é útil? Os eventos de uma interface Swing são processados em uma thread comumente chamada de “event loop”. Os eventos são colocados em fila e processados um após o outro. O evento Ei+1 só pode ser processado quando o evento anterior, Ei, tiver sido totalmente processado. Portanto, é importante que o processamento de um evento seja o mais rápido possível para que a interface gráfica permaneça responsiva. Às vezes, o processamento de um evento pode demorar bastante. Esse é o caso quando esse processamento envolve acessos à rede. Se não quisermos que a interface gráfica fique travada de forma inaceitável para o usuário, é necessário que esses acessos à rede sejam realizados em threads separadas do evento loop, a fim de liberá-lo. Entramos, então, no campo da programação concorrente (várias threads são executadas em paralelo), considerada, com razão, difícil. A biblioteca Rx oferece uma solução simples e elegante para esse problema.

Para simular processamentos demorados, o serviço do exemplo gera seus números aleatórios após um certo tempo de espera, para que possamos observar o comportamento da interface gráfica.

2.2. L'exécutable

O executável do aplicativo de exemplo está na pasta [dvp/executables] dos exemplos:

Existem várias maneiras de executar o arquivo [swing-01], dependendo da configuração do computador utilizado para executá-lo. Pode-se, por exemplo, seguir o processo [1-3]. Obtém-se, então, a seguinte interface gráfica:

• a interface apresenta duas abas [1-2]: uma, [Request], para a solicitação ao serviço de geração de números aleatórios; a outra, [Response], para a exibição dos números recebidos;
• em [3], indica-se quantas solicitações se deseja fazer ao serviço;
• em [4], indica-se o intervalo [a,b] de geração dos números desejados;
• em [5], o número de valores retornados pelo serviço será um número aleatório no intervalo [minCount, maxCount] definido pelo usuário;
• em [6], antes de retornar sua resposta, o serviço aguardará delay milissegundos, sendo que delay é um número aleatório no intervalo [minDelay, maxDelay] definido pelo usuário;
• por padrão, a camada [swing] se comunicará com a interface síncrona do serviço. Para se comunicar com a camada assíncrona, o usuário deverá marcar [7]. Nesse caso, o serviço de geração será executado em threads separadas do evento loop da interface gráfica. A biblioteca Rx dispõe de diversas estratégias para a geração dessas threads. O usuário poderá escolher sua estratégia em [8];
• a geração de números é feita com o botão [9];

• em [10], exibição dos resultados. Explicaremos a estrutura desses resultados;
• em [11], o número de resultados obtidos;
• em [12], o tempo de execução em milissegundos;
• No [13], o usuário tem a possibilidade de cancelar a execução;

Cada resultado tem o seguinte formato:

{"idClient":0,"serviceResponse":{"delay":412,"aleas":[146,115,128,174,159,112,162,127],"executedOn":"RxComputationThreadPool-6"},"observedOn":"AWT-EventQueue-0","requestAt":"02:42:47:708","responseAt":"02:42:52:931"}

• [idClient]: o número da solicitação. Vale lembrar que várias solicitações são enviadas ao serviço de geração;
• [delay]: o tempo de espera, em milissegundos, que o serviço observou antes de enviar seu resultado;
• [aleas]: os números aleatórios retornados pelo serviço;
• [executedOn]: o nome do thread no qual o serviço foi executado;
• [observedOn]: o nome do thread que exibiu o resultado. Com uma interface Swing, esse só pode ser o thread do event loop, neste caso [AWT-EventQueue-0];
• [requestAt]: a hora da solicitação no formato [heures:minutes:secondes:millisecondes];
• [responseAt]: a hora de recebimento dos resultados no mesmo formato;

Apresentaremos agora as partes do código úteis para a compreensão do exemplo.

2.3. A interface síncrona

A camada de serviço [1] apresenta a seguinte interface:

public interface IService {
  // números aleatórios em [a,b]
  // n números são gerados com n aleatórios no intervalo [minCount, maxCount]
  // os números são gerados após um intervalo de espera de delay milissegundos,
  // onde [delay] é um número aleatório no intervalo [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

A resposta [ServiceResponse] é a seguinte:

public class ServiceResponse {

  // tempo de espera do serviço
  private int delay;
  // números aleatórios
  private List<Integer> aleas;
  // thread de execução
  private String executedOn;

  // construtores

  public ServiceResponse(int delay, List<Integer> aleas) {
    executedOn = Thread.currentThread().getName();
    this.delay = delay;
    this.aleas = aleas;
  }

  // getters e setters
...
}

A resposta contém três elementos:

• linha 6: os números aleatórios gerados;
• linha 4: o tempo de espera observado pelo serviço antes de retornar seu resultado;
• linha 8: o thread de execução do serviço;

2.4. A chamada síncrona

Apresentamos agora em detalhes a chamada síncrona [4] que a camada [swing] faz ao serviço [1]:

  private void doGenerateWithService() {
    // início da espera
    beginWaiting();
    try {
      for (int i = 0; i < nbRequests; i++) {
        UiResponse uiResponse = new UiResponse();
        uiResponse.setIdClient(i);
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        uiResponse.setResponseAt();
        model.add(0, jsonMapper.writeValueAsString(uiResponse));
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
      }
    } catch (JsonProcessingException | RuntimeException e) {
      System.out.println(e);
    }
    // fim da espera
    endWaiting();
}

• linhas 5-12: o ciclo de execução das solicitações [nbRequests] feitas pelo usuário;
• linha 8: [service] é a implementação da interface síncrona [IService] apresentada no parágrafo 2.3;
• linha 10: [model] é o modelo exibido pelo componente JList da guia [Response]. Os elementos desse modelo são as cadeias jSON dos seguintes elementos do tipo [UiResponse]:

public class UiResponse {

  // ID do cliente
  private int idClient;
  // resposta do serviço
  private ServiceResponse serviceResponse;
  // nome da thread de observação
  private String observedOn;
  // hora da solicitação
  private String requestAt;
  // hora da resposta
  private String responseAt;

  // construtores

  public UiResponse() {
    observedOn = Thread.currentThread().getName();
    requestAt = getTimeStamp();
  }
  // métodos privados

  private String getTimeStamp() {
    return new SimpleDateFormat("hh:mm:ss:SSS").format(Calendar.getInstance().getTime());
  }

  // getters e setters
...
}

• linha 6: a resposta do serviço de geração de números;
• linha 4: o número da solicitação à qual foi respondido;
• linha 8: o thread de exibição dessa resposta. Como já foi dito, será sempre o thread do event loop;
• linhas 10 e 12: a hora da solicitação e a hora da resposta;

2.5. Testes de chamadas síncronas

Executamos a seguinte configuração:

Obtemos os seguintes resultados na aba [Response]:

• em [1-2], obtivemos efetivamente 10 respostas, conforme solicitado. Elas foram inseridas na primeira posição, na ordem em que chegaram. Percebe-se que foram obtidas na ordem das solicitações;
• todas foram executadas e exibidas na thread do event loop [AWT-EventQueue-0]. As consultas, portanto, foram executadas uma após a outra nessa thread. Não houve consultas simultâneas;
• o que não fica visível aqui é que, durante a execução, a interface gráfica fica congelada. Não há, por exemplo, como acessar a aba [Response] para ver as respostas chegando ou interromper a execução com o botão [Annuler]. Mesmo que esse botão estivesse presente na aba [Request], ele estaria inutilizável. De fato, haveria então dois eventos:
  • o clique no botão [Générer];
  • o clique no botão [Annuler];

O clique no botão [Annuler] só é processado após o término da operação iniciada pelo clique no botão [Générer]. Acabamos de ver que essa operação ocupava o thread do event loop durante toda a execução, impedindo, assim, o processamento do clique no botão [Annuler]. Esse é tipicamente o tipo de situação em que o Rx pode trazer uma melhoria significativa;

2.6. A interface assíncrona e sua implementação

Vamos nos concentrar agora na interface da camada [2] e em sua implementação com o Rx. Ela não será compreensível imediatamente. Queremos apenas destacar a simplicidade do código dessa implementação.

A interface assíncrona é a seguinte:

public interface IRxService {
  // números aleatórios em [a,b]
  // n números são gerados com n aleatórios no intervalo [minCount, maxCount]
  // os números são gerados após um intervalo de espera de delay milissegundos,
  // onde [delay] é um número aleatório no intervalo [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

As diferenças em relação à interface síncrona apresentada no parágrafo 2.3 são as seguintes:

• a classe [UiResponse] apresentada no parágrafo 2.3 agora faz parte dos parâmetros do método [getAleas] (linha 6). A razão para isso é que, como as solicitações agora são executadas em paralelo e o serviço aguarda um tempo aleatório antes de retornar seu resultado, as respostas não nos chegarão na mesma ordem das solicitações. Portanto, passamos o objeto [UiResponse], que contém, entre outras informações, o número da solicitação:

  // ID do cliente (solicitação)
  private int idClient;
  // resposta do serviço
  private ServiceResponse serviceResponse;
  // nome do thread de observação
  private String observedOn;
  // hora da solicitação
  private String requestAt;
  // hora da resposta
  private String responseAt;

• O tipo da resposta do serviço assíncrono é [Observable<UiResponse>]. O tipo [Observable<>] é fornecido pela biblioteca Rx. O resultado do tipo [Observable<UiResponse>] indica que o método [getAleas] fornece um fluxo de valores do tipo [UiResponse], valores esses que são enviados (pushed) um por um para seu observador;

Vejamos agora a implementação dessa interface:

public class RxService implements IRxService {

  // serviço
  private IService service;

  // fabricante
  public RxService(IService service) {
    this.service = service;
  }

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
  }
}

• linhas 7-9: fornece-se ao construtor uma referência à interface síncrona [IService]. É ela que se encarregará de gerar os números aleatórios;
• o observável retornado pelo método [getAleas] é construído pelo método estático [Observable.create]. É esse método que permite construir uma implementação assíncrona a partir de uma implementação síncrona;
• linha 13: o parâmetro do método estático [Observable.create] é, neste caso, uma função lambda que recebe como parâmetro um tipo [Subscriber], que, mais uma vez, é um tipo Rx. Um [Subscriber] é um objeto que se inscreve em um fluxo de observáveis, ou seja, um fluxo de dados fornecidos de forma assíncrona. Aqui, utilizam-se três métodos desse assinante:
  • [Subscriber.onNext] para enviar um dado a ele (linha 16);
  • [Subscriber.onError] para enviar uma exceção a ele (linha 18);
  • [Subscriber.onCompleted] para indicar ao assinante que o fluxo de dados foi concluído (linha 20);

Pode haver vários assinantes para um mesmo observável. Aqui, teremos apenas um assinante que se inscreve em um fluxo de um único dado, aquele produzido nas linhas 15-16. O dado é produzido pela implementação síncrona do serviço (linha 15) e entregue ao assinante (linha 16).

Mesmo que tudo isso provavelmente ainda pareça obscuro, não dá para deixar de ficar impressionado com a extrema concisão dessa implementação assíncrona do serviço.

2.7. A chamada assíncrona

Detalhamos agora a chamada síncrona [5] que a camada [swing] faz ao serviço [2]:

private void doGenerateWithRxService() {
        // início da espera
        beginWaiting();
        // solicitação de números aleatórios
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // agendador
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

• linhas 6-10: execução das consultas [nbRequests] solicitadas pelo usuário;
• linhas 7-8: preparação do objeto [UiResponse] necessário para o método [getAleas] do serviço assíncrono (linha 13). Trata-se principalmente de registrar o número [idClient] da solicitação;
• linha 13: o método [getAleas] do serviço assíncrono é chamado. Ele retorna um objeto [Observable<UiResponse>]. Essa chamada ainda não invoca o serviço síncrono. Voltemos ao código do [getAleas] assíncrono:

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

O código das linhas 4 a 11, que chamará o serviço síncrono, só é executado quando um assinante se registra. Enquanto não houver assinantes, esse código não é executado.

Voltemos ao código do método [doGenerateWithRxService]:

• linha 5: cria-se um observável vazio (nada é observado);
• linha 13: cria-se um observável cujo fluxo será a fusão dos fluxos assíncronos [nbRequests] associados às solicitações [nbRequests]. Isso é obtido com o método [Observable.mergeWith], que permite fundir dois fluxos assíncronos. Na terminologia Rx, [mergeWith] é chamado de operador de fluxo. Esses operadores têm a particularidade de que o resultado da operação é, na maioria das vezes, novamente um [Observable]. No final, após a linha 17, a variável [observables] designa um único fluxo constituído pelas respostas assíncronas [nbRequests] fornecidas pelo serviço assíncrono;
• linha 13: a operação de fusão poderia ter sido escrita da seguinte forma:

observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse));

mas escrevemos:

observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));

Utilizamos aqui o operador [subscribeOn] sobre o observável [rxService.getAleas]. Como costuma acontecer, o resultado é novamente um observável. O operador [subscribeOn] permite especificar que o observável deve ser executado em um thread fornecido por um [Scheduler]. Existem vários [Scheduler] possíveis, adequados a diferentes situações. Na interface gráfica, propusemos vários deles para observar os efeitos de cada um:

Isso gera o seguinte código:

    private void doGenerateWithRxService() {
        // início da espera
        beginWaiting();
        // solicitação de números aleatórios
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // agendador
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
            case 1:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.computation()));
                break;
            case 2:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.newThread()));
                break;
            case 3:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.trampoline()));
                break;
            case 4:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.immediate()));
                break;
            }
        }
...
}

Voltemos ao código das linhas 12 a 14. O agendador [Schedulers.io()] atribui um novo thread a cada observável. Se seguirmos o código:

• linha 5: temos um observável vazio;
• linha 13, iteração 1: observables é a lista [observable0/thread0] (observável observable0 executado na thread thread0);
• linha 13, iteração 2: observables é a lista [observable0/thread0, observable1/thread1];
• etc...

No final, após a linha 28, temos um observável resultante da fusão dos observáveis [nbRequests] que são executados em threads diferentes de [nbRequests]. Nem todos os agendadores funcionam dessa maneira, como veremos nos testes.

Vamos continuar a análise do código de chamada do serviço assíncrono:

private void doGenerateWithRxService() {
        // início da espera
        beginWaiting();
        // solicitação de números aleatórios
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // observador
        observables = observables.observeOn(SwingScheduler.getInstance());
        // executa-se esses observáveis
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

• vimos que, ao chegarmos à linha 10, temos um único observável, resultado da fusão de [nbRequests] observáveis que podem ou não ser executados em [nbRequests] threads diferentes, dependendo do agendador escolhido pelo usuário;
• linha 10: o operador [observeOn] permite especificar em qual thread queremos recuperar os dados provenientes do observável, neste caso, os objetos do tipo [UiResponse]. Em uma interface Swing, não há escolha. Qualquer atualização da interface deve ser feita na thread do evento loop. Aqui, os dados do observável serão exibidos em um componente Swing JList. A thread [SwingScheduler.getInstance()] representa a thread do loop de eventos. A classe [SwingScheduler] não provém da biblioteca RxJava, mas sim da biblioteca derivada RxSwing;
• ao chegar à linha 12, o serviço síncrono ainda não foi chamado, pois o observável da linha 10 ainda não possui nenhum assinante. As linhas 12 a 17 atribuem um a ele, por meio do operador [subscribe]. Os parâmetros desse operador são, neste caso, três funções lambda:
  • a primeira, [uiResponse -> {updateUi(uiResponse);}], aceita como parâmetro um dos objetos [UiResponse] produzidos pelo observável. Vale lembrar que, neste caso, teremos [nbRequests] objetos desse tipo. O método associado, updateUi neste caso, deve utilizar esse resultado;
  • A segunda [th -> {System.out.println(th);doCancel();}] aceita como parâmetro um tipo [Throwable], neste caso, uma exceção que ocorreu durante a execução do observável. O método associado deve utilizar essa informação. Aqui, exibimos a informação no console (linha 15) e cancelamos a execução, o que resultará na atualização de alguns elementos da interface gráfica;
  • o terceiro [this::doCancel] é chamado quando o observável sinaliza que não tem mais dados a transmitir. Aqui, o observável é a união dos observáveis [nbRequests]. O observável resultante indicará que terminou quando todos os observáveis que o compõem tiverem, por sua vez, sinalizado que concluíram seu trabalho. Portanto, quando essa terceira função lambda é executada, todos os dados já foram recebidos. O método local [doCancel] atualiza a interface gráfica para refletir o fato de que a execução foi concluída;

A variável [subscriptions] é definida da seguinte forma:

    // as assinaturas dos observáveis
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

O tipo [Subscription] representa uma assinatura, ou seja, a ligação entre um assinante [Subscriber] e o que ele observa [Observable]. Utilizamos aqui uma lista de assinaturas, embora, neste exemplo, haja apenas uma. O método local [doCancel], executado quando o observável sinaliza que não tem mais dados a transmitir, é o seguinte:

    @Override
    protected void doCancel() {
        // fim da espera
        endWaiting();
        // no caso de assinaturas
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

• a linha 7 cancela a assinatura de todos os assinantes do observável;

A partir dessa explicação resumida, podemos destacar os seguintes pontos-chave:

• o tipo [Observable] designa um fluxo de valores, valores esses que são enviados um a um para assinantes ou observadores;
• o tipo [Subscriber] designa um assinante do tipo [Observable];
• o tipo [Subscription] designa uma assinatura, ou seja, a ligação entre um [Subscriber] e um [Observable];
• o tipo [Observable] admite operadores [mergeWith, empty, subscribeOn, observeOn, ...], que, em sua maioria, produzem observáveis. Esses operadores servem para configurar o observável antes de sua execução:
  • o que se deseja observar;
  • o thread no qual o observável é executado;
  • o thread no qual o assinante recebe os dados do observável;
• distinguem-se dois tipos de observáveis: os [froid / cold] e os [chaud / hot]. Um observável frio é executado integralmente a cada novo assinante. Se cada execução produzir os mesmos dados, cada novo assinante receberá os mesmos dados que o anterior. Um observável quente geralmente produz dados de forma contínua. Quando um assinante se inscreve, ele recebe os dados emitidos a partir do momento da sua inscrição. Ele não recebe os dados que possam ter sido emitidos anteriormente. No nosso exemplo, o observável é frio: ele é totalmente reexecutado a cada novo assinante. O que é realmente executado no nosso exemplo? Para descobrir isso, precisamos voltar à definição do observável observado:

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

A cada novo assinante, a função lambda, parâmetro do método [Observable.create] (linha 3), é reexecutada. Portanto, são as linhas 4 a 11 que são executadas para cada novo assinante [subscriber];

2.8. Testes de chamadas assíncronas

Começamos mostrando o efeito dos diferentes agendadores propostos. Para isso, utilizamos os seguintes parâmetros:

Definimos valores pequenos em [1-2] para que, caso as consultas sejam executadas na mesma thread, a espera não seja muito longa.

2.8.1. com o agendador [Schedulers.io]

É possível observar os seguintes pontos:

• as respostas são recebidas em uma ordem diferente da das consultas (ver idClient);
• cada solicitação foi executada em um thread diferente;
• a interface gráfica, desta vez, não está mais travada:
  • é possível alternar entre as abas;
  • vemos os dados chegando;
  • não dá tempo de ver o botão [Annuler] porque a execução é muito rápida. Vamos destacá-lo em outro teste;

2.8.2. com o agendador [Schedulers.computation]

É possível observar os seguintes pontos:

• recebemos as respostas em uma ordem diferente daquela das consultas (veja idClient);
• as consultas foram executadas em 8 threads;
• o thread nº 3 foi utilizado para as consultas 8 e 0;
• o thread nº 4 foi utilizado para as consultas 9 e 1;
• as demais consultas tiveram, cada uma, um thread diferente;

O agendador [Schedulers.computation] utiliza tantos threads quantos forem os núcleos da máquina utilizada. Essa informação é obtida pela expressão [Runtime.getRuntime().availableProcessors()].

2.8.3. com o agendador [Schedulers.newThread]

O funcionamento é análogo ao do agendador [Schedulers.io].

2.8.4. com os agendadores [Schedulers.trampoline, Schedulers.immediate]

Temos um funcionamento síncrono. Todas as consultas são executadas na thread do event loop. Não se deve generalizar esse resultado, mas sim afirmar que, simplesmente neste exemplo específico, os dois agendadores funcionaram de forma síncrona.

2.9. Casos extremos

Vamos trabalhar com os agendadores que permitem um funcionamento assíncrono neste exemplo. Primeiramente, aumentamos o número de solicitações para 100 com o agendador [Schedulers.computation], que trabalha aqui com 8 threads. Obtemos o seguinte resultado:

• no [1], o botão [Annuler] está presente e pode ser utilizado (funcionamento assíncrono);

Agora, vamos deixar a execução seguir até o fim:

No [2], vemos que a execução das 100 consultas levou cerca de 4 segundos (em 8 threads).

Agora, vamos fazer essas mesmas 100 consultas com o agendador [Schedulers.newThread], que executa cada consulta em um thread separado:

No [1], vemos que a execução das 100 consultas (em 100 threads) levou meio segundo. Portanto, é significativamente mais rápido do que com o agendador [Schedulers.computation].

Agora, vamos fazer 800 solicitações nas mesmas condições, ainda com o agendador [Schedulers.newThread]. Obtemos os seguintes resultados:

As 800 consultas são executadas em aproximadamente 1 segundo.

Quando aumentamos esse número (para mais de 2.500 solicitações na minha máquina — executadas em 1,5 s — esse número, é claro, depende muito do ambiente de trabalho no momento da execução), acabamos obtendo a seguinte exceção:

Portanto, ocorre um estouro de pilha. Os testes mostram que o funcionamento do agendador [Schedulers.newThread] não é determinístico. É possível obter a exceção anterior, realizar novos testes, voltar à configuração que provocou a exceção e não obtê-la mais.

2.10. Conclusion

Mostramos um exemplo de uso da biblioteca Rx. Vamos resumir o que aprendemos:

Partimos da seguinte arquitetura:

• em [4], a camada [swing] fazia chamadas síncronas para a camada [service];
• na camada [5], a camada [swing] fazia chamadas assíncronas para a camada [rxService], que, por sua vez, chamava de forma síncrona a camada [service] por meio da camada [6];

A primeira coisa que observamos foi que a biblioteca Rx permitia criar facilmente a interface assíncrona [rxService] a partir da interface síncrona [service] (ver parágrafo 2.4). Essa é uma lição importante, pois significa que é possível fazer a transição de uma aplicação síncrona para uma assíncrona com facilidade.

Na camada [swing], foram criados dois métodos distintos:

• um para realizar chamadas síncronas ao serviço (ver parágrafo 2.4);
• o outro para fazer chamadas assíncronas (ver parágrafo 2.7);

A implementação das chamadas assíncronas revelou-se significativamente mais complexa do que a das chamadas síncronas. No entanto, quem já trabalhou com programação concorrente envolvendo várias threads a serem sincronizadas perceberá que a solução Rx é mais simples de implementar e evita todos os problemas de sincronização e comunicação entre threads, que são questões complexas. Durante essa implementação, destacamos os seguintes pontos importantes:

• o tipo [Observable] designa um fluxo de eventos (valores) que podem ser (mas não necessariamente) assíncronos e que podem ser observados;
• o tipo [Subscriber] designa um assinante de um tipo [Observable];
• o tipo [Subscription] designa uma assinatura, ou seja, a ligação entre um [Subscriber] e um [Observable];
• o tipo [Observable] admite operadores [mergeWith, empty, subscribeOn, observeOn, ...], que, em sua maioria, produzem observáveis. Esses operadores servem para configurar o observável antes de sua execução:
  • o que se deseja observar;
  • o thread no qual o observável é executado;
  • o thread no qual o assinante recebe os dados do observável;
• distinguem-se dois tipos de observáveis: os [froid / cold] e os [chaud / hot]. Um observável frio é executado integralmente a cada novo assinante. Se cada execução produzir os mesmos dados, cada novo assinante receberá os mesmos dados que o anterior. Um observável quente geralmente produz dados de forma contínua. Quando um assinante se inscreve, ele recebe os dados emitidos a partir do momento da sua inscrição. Ele não recebe os dados que possam ter sido emitidos anteriormente. No nosso exemplo, o observável é frio: ele é totalmente reexecutado a cada novo assinante.

Agora que vimos um exemplo que nos mostrou a utilidade da biblioteca Rx, vamos apresentá-la com mais detalhes.

A biblioteca Rx possui vários métodos com parâmetros genéricos em suas assinaturas. Faremos uma breve revisão dessas assinaturas (parágrafo 3). Os parâmetros desses métodos são, na maioria das vezes, interfaces funcionais (Java 8), ou seja, interfaces que possuem apenas um único método. Os parâmetros efetivos devem, portanto, ser instâncias dessas interfaces. Antes do Java 8, era comum implementar uma interface por meio de uma classe anônima. Com o Java 8, se a interface for funcional, é mais conciso implementá-la com uma função lambda. Apresentaremos, portanto, essas funções (parágrafo 4). Depois disso, apresentaremos a classe [Stream] (parágrafo 5), que permite processar coleções Java com funções lambda. Essa classe é interessante porque a classe [Observable], de RxJava, herda dela:

• certos métodos;
• a mesma forma de encadear métodos entre si para processar um mesmo observável;

Em seguida, apresentaremos as interfaces funcionais específicas da biblioteca RxJava (parágrafo 6). Continuaremos com os principais elementos da biblioteca Rx [Observable, Subscriber, Subscription, opérateurs] (parágrafo 7). A classe [Observable] possui dezenas de operadores que, por sua vez, são sobrecarregados várias vezes. Isso gera, inicialmente, uma grande complexidade, pois esses operadores e suas sobrecargas às vezes diferem apenas em um detalhe, e é difícil, sem experiência, saber qual operador utilizar. Apresentaremos apenas um número limitado de operadores e, na maioria das vezes, ignoraremos suas sobrecargas.

Toda a parte anterior será realizada com a biblioteca RxJava em aplicativos de console simples. Assim que a biblioteca RxJava for adquirida, a utilizaremos em dois tipos de aplicativos gráficos:

• no parágrafo 8, voltaremos à aplicação Swing de exemplo para detalhá-la mais a fundo. Utilizaremos, então, a biblioteca RxSwing;
• no parágrafo 9, criaremos um aplicativo Android com a biblioteca RxAndroid;

Quando tudo isso estiver concluído, o leitor terá as ferramentas necessárias para seguir seu próprio caminho. Provavelmente levará algum tempo até que ele consiga usar a biblioteca Rx de forma intuitiva. Achei essa biblioteca particularmente interessante. No entanto, achei-a complexa de entender e o tempo de aprendizado foi longo. Espero que este documento ajude a encurtar esse tempo para o leitor. Parece-me que vale a pena o esforço.