Projetando, implementando e usando Reactive APIs

Nos últimos dois anos, houve um grande crescimento da programação reativa no mundo Java. Este é o sucesso dos desenvolvedores de NodeJS que utilizam APIs que não bloqueiam a expansão dos microservices indutores de latência, ou simplesmente o desejo de utilizar recursos computacionais de forma mais eficiente, muitos desenvolvedores estão começando a olhar para a programação reativa como um modelo de programação viável.

Felizmente, os desenvolvedores de Java não são prejudicados pela escolha quando se trata de estruturas reativas e como usá-las adequadamente. Não há muitas maneiras "erradas" de escrever um código reativo, mas nisso reside o atrito; Existem poucas maneiras "corretas" de escrever um código reativo.

Neste artigo, o objetivo foi passar algumas sugestões sobre como você deve escrever um código reativo. Essas opiniões vêm de anos de experiência no desenvolvimento de uma API reativa em larga escala e, embora possam não estar todas de acordo para você, esperamos que elas lhe deem alguma direção à medida que começa sua jornada reativa.

Os exemplos neste artigo são todos provenientes do Cloud Foundry Java Client. Este projeto usa o Project Reactor para sua estrutura reativa. O Reactor foi escolhido para o Java Client devido à sua estreita integração com a equipe Spring, mas todos os conceitos discutidos aplicam-se a outros frameworks reativos como, por exemplo o RxJava. Embora seja útil ter alguma compreensão da Cloud Foundry, não é de extrema importância. Os exemplos têm uma indicação auto-explicativa que irão guiá-los através do conceito relativo que cada um está demonstrando.

A programação reativa é um assunto vasto e está muito além do alcance deste artigo, mas para nossos propósitos, vamos defini-la amplamente, como uma maneira de desenvolver sistemas orientados a eventos de uma maneira mais fluente do que com um estilo de programação imperativa.

Muitas das APIs projetadas para esses comportamentos (threads, NIO callbacks, etc.) não são consideradas de fácil manuseio e de forma correta e confiável e, em muitos casos, para utilizar essas APIs é necessário uma quantidade razoável de gerenciamento explícito no código do aplicativo. A promessa de um quadro reativo é que essas preocupações podem ser tratadas nos bastidores, permitindo que o desenvolvedor escreva um código que se centra principalmente na funcionalidade do aplicativo.

Devo usar a programação reativa?

A primeira pergunta a se fazer ao projetar uma API reativa é se você deseja mesmo uma API reativa! As APIs reativas não são a escolha correta para todas as questões. Existem desvantagens demonstráveis para a programação reativa (a depuração é a maior hoje em dia, mas ambas os frameworks e IDEs estão trabalhando nelas). Em vez disso, você escolhe uma API reativa quando o valor obtido supera significativamente as desvantagens. Ao fazer este julgamento, existem alguns padrões óbvios para os quais a programação reativa é um bom ajuste.

Redes

As solicitações de rede envolvem altas latências, e aguardar essas respostas para retornar é muitas vezes o maior desperdício de recursos em um sistema. Em uma aplicação não-reativa, esses pedidos de espera geralmente bloqueiam threads e consomem memória da pilha, esperando ociosamente uma resposta chegar. Falhas remotas e tempos limite geralmente não são tratados de forma sistemática e explícita porque as APIs fornecidas não facilitam a tarefa. Finalmente, as cargas úteis de chamadas remotas são muitas vezes de um tamanho desconhecido e ilimitado, levando ao esgotamento da memória do heap. A programação reativa, combinada com o IO não bloqueante, aborda esses tipos de problemas, pois oferece uma API clara e explícita para cada um.

Operações de alta concorrência

A coordenação de operações são altamente concorrentes, como os pedidos de rede ou cálculos paralisantes de CPU. Estruturas reativas permitem ao mesmo tempo o gerenciamento explícito de encadeamento excel quando são deixados para gerenciar o encadeamento automático. Os operadores gostam de .flatMap() para paralisar os comportamentos de forma transparente, maximizando o uso dos recursos disponíveis.

Aplicações em grande escala

O modelo de servlet em um segmento por conexão nos serviu por muito bem, mas já está ultrapassado. Com microservices começamos a ver aplicações amplamente dimensionadas (25, 50 até 100 instâncias de uma única aplicação sem estado) para lidar com cargas de conexão mesmo quando o uso da CPU está ocioso. Escolhendo IO não bloqueante, com programação reativa torna-se aceitável a quebra dessa ligação e faz um uso muito mais eficiente dos recursos disponíveis. Para ser claro, a vantagem passa a ser muitas vezes surpreendente. Geralmente, muitas outras instâncias de um aplicativo são criadas no Tomcat com centenas ou milhares de threads para lidar com a mesma carga do mesmo aplicativo criado no Netty com oito threads.

O que uma API reativa deve retornar?

Se você respondeu a primeira pergunta e determinou que seu aplicativo se beneficiará de uma API reativa, é hora de projetar a sua API. um bom lugar para começar é decidir quais são os tipos primitivos que sua API reativa deve retornar.

Todas as estruturas reativas no mundo Java (incluindo o fluxo do Java 9) estão convergentes na Especificação dos Fluxos Reativos. Esta especificação define uma API de interoperabilidade de baixo nível, mas não é considerada uma estrutura reativa (ou seja, não especifica os operadores disponíveis para fluxo)

Existem dois diferentes tipos do Project Reactor, em que a programação reativa se baseia. O tipo Flux <T> representa 0 a N valores que fluem através do sistema. O tipo Mono <T> representa os valores 0 a 1. Dentro do Java Client, usamos o Mono quase que exclusivamente, pois ele mapeia claramente para um modelo de solicitação único e resposta única.

Flux<Application> listApplications() {...}

Flux<String> listApplicationNames() {
  return listApplications()
    .map(Application::getName);
}

void printApplicationName() {
  listApplicationNames()
    .subscribe(System.out::println);
}

Neste exemplo, o método listApplications() executa uma chamada de rede e retorna um Flux de 0 para N instâncias da aplicação. Em seguida, usamos o operador .map() para transformar cada aplicação em uma cadeia de caracteres de seu nome. Esse fluxo de nomes de aplicativos é então consumido e impresso para o console.

Flux<Application> listApplications() {...}

Mono<List<String>> listApplicationNames() {
  return listApplications()
    .map(Application::getName)
    .collectList();
}

Mono<Boolean> doesApplicationExist(String name) {
  return listApplicationNames()
    .map(names -> names.contains(name));
}

Os Monos não tem um fluxo da mesma forma que os Fluxs fazem, mas como eles são conceitualmente um fluxo de um elemento, os operadores que usamos geralmente têm o mesmo nome. Neste exemplo, além de mapear para um fluxo de nomes de aplicativos, nós coletamos esses nomes em uma única lista. O Mono que contém essa lista pode então ser transformado, neste caso, em um booleano indicando se um nome está contido nele. Pode não ser intuitivo, mas é comum retornar um Mono de coleções (por exemplo, Mono<List<String>>) se o item com o qual você está trabalhando é logicamente uma coleção de itens em vez de um fluxo deles.

Ao contrário das APIs imperativas, o vazio não é um tipo de retorno reativo apropriado. Em vez disso, cada método deve retornar um Flux ou um Mono. Isso provavelmente parece estranho (ainda há comportamentos que não tem nada para retornar!), mas é um resultado de operações básicas de fluxo reativo. A execução do código que chama APIs reativas (por exemplo, .flatMap() .Map() …) está construindo uma estrutura para que os dados fluam, mas não transformam realmente os dados. Somente no final, quando chamado .subscribe(), os dados começam a se mover através do fluxo, sendo transformados pelas operações à medida que são executados. Esta execução preguiçosa é por que a programação reativa é construída em cima de lambdas e porque os tipos são sempre retornados; sempre deve haver algo para .subscribe().

void delete(String id) {
  this.restTemplate.delete(URI, id);
}

public void cleanup(String[] args) {
  delete("test-id");
}

O exemplo imperativo e bloqueador acima pode retornar vazio à medida que a execução da chamada de rede começe e não retorna até que a resposta seja recebida.

Mono<Void> delete(String id) {
  return this.httpClient.delete(URI, id);
}

public void cleanup(String[] args) {
  CountDownLatch latch = new CountDownLatch(1);

  delete("test-id")
    .subscribe(n -> {}, Throwable::printStackTrace, () -> latch::countDown);

  latch.await();
}

Neste exemplo, a chamada de rede não começa até que seja chamado o .subscribe(), depois que o delete() retorna, porque é a estrutura para fazer essa chamada, não o resultado da chamada em si. Nesse caso, tempo o equivalente a um tipo de retorno vazio usando um Mono<Void> que retorna 0 itens, sinalizando o onComplete() somente após a resposta ter sido recebida.

Escopo dos métodos

Depois de ter decidido sobre o que suas APIs precisam retornar é necessário examinar o que cada um de seus métodos (tanto API quanto implementação) fará. No Java Client, descobrimos que os métodos de design são pequenos e reutilizáveis. Ele permite que cada um desses métodos sejam facilmente compostos em operações maiores. Também permite que eles sejam combinados de forma mais flexível em operações paralelas ou sequenciais. Como um bônus, ele também torna os fluxos potencialmente complexos muito mais legíveis.

Mono<ListApplicationsResponse> getPage(int page) {
  return this.client.applicationsV2()
    .list(ListApplicationsRequest.builder()
      .page(page)
      .build());
}

void getResources() {
  getPage(1)
    .flatMapMany(response -> Flux.range(2, response.getTotalPages() - 1)
      .flatMap(page -> getPage(page))
      .startWith(response))
    .subscribe(System.out::println);
}

Este exemplo demonstra como chamamos uma API paginada. A primeira requisição ao getPage() recupera a primeira página de resultados. Incluído na primeira página de resultados está o número total de páginas que precisamos recuperar para obter o resultado completo. Como o método getPage() é pequeno, reutilizável e sem efeitos colaterais, podemos então utilizar o método e chamar a segunda página através do totalPages em paralelo!

Coordenação Sequencial e Paralela

Hoje em dia quase todas as melhorias de desempenho significativas vêm da concorrência. Nós sabemos disso e, no entanto, muitos sistemas são apenas concorrentes em relação às conexões recebidas ou não são concorrentes. Grande parte dessa situação pode ser rastreada, pois implementar um sistema altamente concorrente é difícil e propenso a erros. Um dos principais benefícios da programação reativa é que você define relacionamentos sequenciais e paralelos entre as operações e deixe a estrutura determinar a maneira ideal de usar os recursos disponíveis.

Examinando novamente o exemplo anterior, a primeira chamada para getPage() é garantida para ocorrer sequencialmente antes das chamadas posteriores para cada página adicional. Além disso, uma vez que essas chamadas subsequentes para getPage() acontecem em um .flatMapMany(), a estrutura é responsável por organizar de forma otimizada sua execução e juntar os resultados de volta, propagando quaisquer erros que possam ter ocorrido

Lógica Condicional

Diferente de uma programação imperativa, os erros são considerados valores na programação reativa. Isso significa que eles passam pelas operações no fluxo. Esses erros podem ser passados até os consumidores, ou os fluxos podem mudar de comportamento com base neles. Essa alteração de comportamento pode se manifestar como a transformação de erros ou a geração de novos resultados com base em um erro.

public Mono<AppStatsResponse> getApplication(GetAppRequest request) {
  return client.applications()
    .statistics(AppStatsRequest.builder()
      .applicationId(request.id())
      .build())
    .onErrorResume(ExceptionUtils.statusCode(APP_STOPPED_ERROR),
      t -> Mono.just(AppStatsResponse.builder().build()));
}

Neste exemplo, um pedido é feito para obter as estatísticas de um aplicativo em execução. Se tudo funcionar como esperado, a resposta é repassada ao consumidor. No entanto, se um erro for recebido (com um código de status específico), uma resposta vazia será retornada. O consumidor nunca vê o erro e a execução prossegue com um valor padrão, como se o erro nunca tivesse ocorrido.

Em uma discussão anterior, foi válido para um fluxo completo sem o envio de nenhum item. Este é frequentemente o equivalente a retornar nulo (dos quais um tipo de retorno vazio é um caso especial). Como o caso do erro, esta conclusão sem qualquer item pode ser passada para consumidores, ou os fluxos podem mudar o comportamento com base no erro.

public Flux<GetDomainsResponse> getDomains(GetDomainsRequest request) {
  return requestPrivateDomains(request.getId())
    .switchIfEmpty(requestSharedDomains(request.getId()));
}

Neste exemplo, getDomains() retorna um domínio que pode estar em um dos dois diferentes buckets. Primeiro, os domínios privados são pesquisados, e se isso for concluído com sucesso, embora sem resultados, os domínios compartilhados são pesquisados.

public Mono<String> getDomainId(GetDomainIdRequest request) {
  return getPrivateDomainId(request.getName())
    .switchIfEmpty(getSharedDomainId(request.getName()))
    .switchIfEmpty(ExceptionUtils.illegalState(
      "Domain %s not found", request.getName()));
}

Também pode ser o caso de nenhum item indicar uma condição de erro. Neste exemplo, se nenhum domínio privado ou compartilhado pode ser encontrado, uma nova IllegalStateException é gerada e transmitida ao consumidor.

Às vezes você deseja tomar decisões não com bases nos erros ou no vazio, mas nos próprios valores. Embora seja possível implementar esta lógica usando operadores, muitas vezes se revela mais complexo do que imagina. Neste caso, você deve usar declarações condicionais imperativas.

public Mono<String> getDomainId(String domain, String organizationId) {
  return Mono.just(domain)
    .filter(d -> d == null)
    .then(getSharedDomainIds()
      .switchIfEmpty(getPrivateDomainIds(organizationId))
      .next()  // select first returned
      .switchIfEmpty(ExceptionUtils.illegalState("Domain not found")))
    .switchIfEmpty(getPrivateDomainId(domain, organizationId)
      .switchIfEmpty(getSharedDomainId(domain))
      .switchIfEmpty(
          ExceptionUtils.illegalState("Domain %s not found", domain)));
}

Este exemplo retorna o id de um determinado nome de domínio, dentro de uma determinada organização (um recipiente hierárquico). Há um detalhe aqui, no entanto, se o domínio for nulo, o nome do domínio explícito é pesquisado e seu ID é retornado. Se você achar esse código confuso, não se desespere, nós também achamos!

public Mono<String> getDomainId(String domain, String organizationId) {
  if (domain == null) {
    return getSharedDomainIds()
      .switchIfEmpty(getPrivateDomainIds(organizationId))
      .next()
      .switchIfEmpty(ExceptionUtils.illegalState("Domain not found"));
  } else {
    return getPrivateDomainId(domain, organizationId)
      .switchIfEmpty(getSharedDomainId(domain))
      .switchIfEmpty(
          ExceptionUtils.illegalState("Domain %s not found", domain));
    }
}

Este exemplo é equivalente, mas usa instruções condicionais imperativas. Muito mais compreensível, você não concordaria?

Testes

Na na prática, a maioria dos fluxos são assíncronos. Isso é problemático para se testar, porque as estruturas de testes são agressivamente síncronas, registrando, passando ou falhando muito antes dos resultados assíncronos serem retornados. Para compensar isso, é necessário bloquear o segmento principal até que os resultados sejam retornados e, em seguida, mover esses resultados para o segmento principal das asserções.

@Test
public void noLatch() {
  Mono.just("alpha")
    .subscribeOn(Schedulers.single())
    .subscribe(s -> assertEquals("bravo", s));
}

Este exemplo, que emite uma String em um segmento não-principal, passa desapercebidamente. A causa raiz desse teste que tem retorno positivo quando não deve ter, é que o método noLatch será concluído sem lançar um AssertionError.

@Test
public void latch() throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);
  AtomicReference<String> actual = new AtomicReference<>();

  Mono.just("alpha")
    .subscribeOn(Schedulers.single())
    .subscribe(actual::set, t -> latch.countDown(), latch::countDown);

  latch.await();
  assertEquals("bravo", actual.get());
}

Este exemplo, embora seja obsceno, usa um CountDownLatch para garantir que o método latch() não retorne até que o fluxo tenha sido concluído. Uma vez que o latch é liberado, uma afirmação é feita no thread principal que lançará um AssertionError, fazendo com que o teste falhe.

Você está perdoado por olhar este código e se recusar a implementar todos os seus testes dessa forma; certamente nós o fizemos. Por sorte, o Reactor fornece uma classe StepVerifier para facilitar o teste.

O teste de um projeto reativo requer mais do que simples bloqueios. Muitas vezes você precisa testar vários valores e erros esperados, garantindo que erros inesperados causem uma falha de teste. O StepVerifier aborda cada um destes.

@Test
public void testMultipleValues() {
  Flux.just("alpha", "bravo")
    .as(StepVerifier::create)
    .expectNext("alpha")
    .expectNext("bravo")
    .expectComplete()
    .verify(Duration.ofSeconds(5));
}

Neste exemplo, o StepVerifier é usado para esperar exatamente que alpha e bravo sejam emitidos e, em seguida, o fluxo completo. Se qualquer um deles não for emitido, um elemento extra será emitido ou um erro será gerado e o teste falhará.

@Test
public void shareFails() {
  this.domains
    .share(ShareDomainRequest.builder()
      .domain("test-domain")
      .organization("test-organization")
      .build())
    .as(StepVerifier::create)
    .consumeErrorWith(t -> assertThat(t)
      .isInstanceOf(IllegalArgumentException.class)
      .hasMessage("Private domain test-domain does not exist"))
    .verify(Duration.ofSeconds(5));
}

Este exemplo usa alguns dos recursos mais avançados do StepVerifier e afirma que não apenas um erro foi sinalizado, mas que é um IllegalArgumentException e que a mensagem corresponde ao esperado.

CountDownLatches

Uma das principais coisas a lembrar sobre estruturas reativas é que elas só podem coordenar suas próprias operações e modelos de encadeamento. Muitos dos ambientes de execução em que a programação reativa terá excessos de threads individuais (por exemplo, recipientes Servlet). Nesses ambientes, a natureza assíncrona da programação reativa não é um problema. No entanto, existem alguns ambientes, como os exemplos de teste acima, em que os processos terminam antes de qualquer segmento individual.

public static void main(String[] args) {
  Mono.just("alpha")
    .delaySubscription(Duration.ofSeconds(1))
    .subscribeOn(Schedulers.single())
    .subscribe(System.out::println);
}

Assim como o método de teste, esse método main() irá terminar antes que o alfa seja emitido.

public static void main(String[] args) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);

  Mono.just("alpha")
    .delaySubscription(Duration.ofSeconds(1))
    .subscribeOn(Schedulers.single())
    .subscribe(System.out::println, t -> latch.countDown(),
               latch::countDown);

    latch.await();
}

E, assim como no exemplo de teste, um CountDownLatch pode garantir que o segmento principal não termine antes do fluxo, independentemente de qual thread ele estará executando.

Bloqueando Fluxos

É bastante comum hoje e em um futuro próximo, interagir com o bloqueio de APIs em programação reativa. Para fazer a ponte entre os dois, pode ser necessário bloquear um enquanto aguarda o resultado do outro. No entanto, alguns dos benefícios da programação reativa, como o uso eficiente de recursos, são perdidos ao fazer a ponte para uma API de bloqueio dessa maneira. Por isso você quer manter o seu código reativo durante o maior tempo possível, bloqueando apenas no último momento. Também vale a pena notar que a conclusão lógica dessa ideia é uma API reativa que pode ser bloqueada, mas uma API de bloqueio que nunca pode ser reativa.

Mono<User> requestUser(String name) {...}

User getUser(String name) {
  return requestUser(name)
    .block();
}

In this example, .block() is used to bridge the single result from a Mono to an imperative return type.

Flux<User> requestUsers() {...}

List<User> listUsers() {
  return requestUsers()
    .collectList()
    .block();
}

Como no exemplo anterior, .block() é usado para unir um resultado para um tipo de retorno imperativo, mas antes que isto possa acontecer, o Flux deve ser coletado em uma única Lista.

Manipulação de erros

Conforme descrito anteriormente, os erros são valores que circulam pelo sistema. Isso significa que nunca há um ponto apropriado para capturar uma exceção. No entanto, é preciso saber como lidar com eles como parte do fluxo. O método subscribe() tem entre 0 e 3 parâmetros que permitem manipular cada item a medida que ele chega, lidar com um erro quando gerado e lidar com a conclusão de um fluxo.

public static void main(String[] args) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);

  Flux.concat(Mono.just("alpha"), Mono.error(new IllegalStateException()))
    .subscribe(System.out::println, t -> {
      t.printStackTrace();
      latch.countDown();
    }, latch::countDown);

  latch.await();
}

Neste exemplo, tanto um valor como um erro são passados para a função. É importante lembrar que ao usar um CountDownLatch, que apenas um onError() ou onComplete() é chamado. Portanto, é necessário liberar uma trava nos casos de erro e sucesso.

Decifrando Referências do Método

Como se pode imaginar, qualquer modelo de programação que se incline para lambdas é suscetível ao "callback hell". No entanto, com um pouco de disciplina e referências de métodos, não é necessário. Algo que qualquer desenvolvedor razoável de Ruby lhe dirá é que pequenos métodos privados (mesmo um one-liners!) são realmente valiosos quando se trata de legibilidade. Se você nomear bem os métodos e usar a sintaxe de referência do método, você pode criar fluxos muito legíveis.

public Flux<ApplicationSummary> list() {
  return Mono
    .zip(this.cloudFoundryClient, this.spaceId)
    .flatMap(function(DefaultApplications::requestSpaceSummary))
    .flatMapMany(DefaultApplications::extractApplications)
    .map(DefaultApplications::toApplicationSummary);
}

Neste exemplo, é possível verificar o fluxo muito bem. Para obter um Flux<ApplicationSummary>, começamos passando o cloudFoundryClient e um SpaceID. Usamos esses métodos para solicitar um resumo do espaço e, em seguida, mapear cada um desses aplicativos para um resumo do espaço. Para qualquer operação individual, não sabemos como ela se comporta. As IDEs facilitam a transição para essas referências de métodos, se necessário, mas esse código não possui a confusão da implementação de cada um deles.

Estilo Livre de Pontos

Ao longo deste artigo, você pôde ter notado que usamos um estilo muito compacto. Isso é chamado de Pointfree Style. O principal benefício é que ele ajude o desenvolvedor a pensar sobre a composição de funções (uma preocupação de alto nível) em vez de arrastar dados (uma preocupação de baixo nível). Nós não diremos que este é um requisito difícil ao escrever programação reativa, mas achamos que a maioria das pessoas prefere (eventualmente).

Mono<Void> deleteApplication(String name) {
  return PaginationUtils
    .requestClientV2Resources(page -> this.client.applicationsV2()
      .list(ListApplicationsRequest.builder()
        .name(name)
        .page(page)
        .build()))
    .single()
    .map(applicationResource -> applicationResource.getMetadata().getId())
    .flatMap(applicationId -> this.client.applicationsV2()
      .delete(DeleteApplicationRequest.builder()
        .applicationId(applicationId)
        .build()));
}

Se você observar este exemplo, é possível imaginar muitos lugares onde as variáveis podem ser atribuídas, os resultados retornados e, em geral, que ele se parece mais com um código imperativo tradicional. No entanto, não é provável que isso aumente a sua legibilidade. Em vez disso, a adição de chaves, ponto-e-vírgulas, sinais de igualdade e declarações de retorno, ao mesmo tempo em que identifica de onde os dados estão vindo e explicando de forma mais notória, é provável que confunda o ponto real do próprio fluxo.

A programação reativa é um assunto vasto, e quase todos estão iniciando nele. Neste ponto, há poucas respostas "erradas" ao escrever um código reativo, mas, ao mesmo tempo, a abundância de opções deixa muitos desenvolvedores confusos quanto a melhor prática. Esperamos que nossas opiniões, nascidas de um projeto em grande escala, o ajudem na sua jornada reativa e que nós o encorajamos a dirigir o estado da arte experimentando e contribuindo com suas descobertas de volta para a comunidade.

Sobre os autores

Ben Hale é o líder da equipe de experiência da Pivotal’s Java Cloud Foundry e é responsável pelo ecossistema em torno dos aplicativos Java executados no Cloud Foundry.

Paul Harris é o principal desenvolvedor da Pivotal’s Cloud Foundry Java Client e é responsável por ativar os aplicativos Java que orquestram e gerenciam o Cloud Foundry