Observável vs Flowable rxJava2

127

Eu estive olhando para o novo rx java 2 e não tenho muita certeza de entender a idéia de backpressuremais ...

Estou ciente de que temos Observableque não tem backpressuresuporte eFlowable tem.

Assim, com base no exemplo, digamos que eu tenho flowablecom interval:

        Flowable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    // do smth
                }
            });

Isso vai falhar após cerca de 128 valores, e isso é bastante óbvio que estou consumindo mais devagar do que recebendo itens.

Mas então nós temos o mesmo com Observable

     Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    // do smth
                }
            });

Isso não trava, mesmo quando eu coloco algum atraso no consumo, ainda funciona. Para fazer o Flowabletrabalho, digamos que eu coloque o onBackpressureDropoperador, a falha desapareceu, mas nem todos os valores são emitidos.

Portanto, a pergunta básica que não consigo encontrar resposta atualmente em minha cabeça é por que devo me preocupar backpressurequando posso usar o plain Observableainda receber todos os valores sem gerenciar o buffer? Ou talvez do outro lado, quais vantagens backpressureme dão em favor do gerenciamento e manuseio do consumidor?

user2141889
fonte

Respostas:

122

O que a contrapressão manifesta na prática são buffers limitados, Flowable.observeOnpossui um buffer de 128 elementos que é drenado o mais rápido que o fluxo de fluxo pode suportar. Você pode aumentar esse tamanho do buffer individualmente para lidar com a origem explosiva e todas as práticas de gerenciamento de contrapressão ainda se aplicam a partir da 1.x.Observable.observeOnpossui um buffer ilimitado que continua coletando os elementos e seu aplicativo pode ficar sem memória.

Você pode usar, Observablepor exemplo:

  • manipulação de eventos da GUI
  • trabalhando com sequências curtas (menos de 1000 elementos no total)

Você pode usar, Flowablepor exemplo:

  • fontes frias e não programadas
  • gerador como fontes
  • acessadores de rede e banco de dados
akarnokd
fonte
Uma vez que este surgiu em outra pergunta - É verdade que os tipos mais restritas, como Maybe, Singlee Completablepode sempre ser usado em vez de Flowablequando eles são semanticamente apropriado?
David.mihola
1
Sim, Maybe, Single, e Completablesão muito pequeno demais para ter qualquer necessidade do conceito de contrapressão. Não há chance de um produtor emitir itens mais rapidamente do que pode ser consumido, uma vez que itens de 1 a 1 serão produzidos ou consumidos.
AndrewF 2/06
Talvez eu não esteja certo, mas para mim, exemplos de Flowable e Observable devem ser trocados.
Yura Galavay 08/02/19
Acho que na pergunta ele está perdendo a estratégia de contrapressão que precisamos fornecer ao Flowable, o que explica por que a exceção de contrapressão ausente é lançada, também explica por que essa exceção desaparece depois que ele aplicou .onBackpressureDrop (). E para o Observable, como não possui essa estratégia e não pode ser fornecida, simplesmente falhará mais tarde devido à OOM
Haomin
110

Contrapressão é quando o observável (editor) está criando mais eventos do que o assinante pode suportar. Assim, você pode fazer com que os assinantes faltem aos eventos ou uma enorme fila de eventos, o que acaba levando à falta de memória. Flowableleva em consideração a contrapressão. Observablenão. É isso aí.

isso me lembra um funil que quando transborda muito líquido. O Flowable pode ajudar a não fazer isso acontecer:

com tremenda contrapressão:

insira a descrição da imagem aqui

mas com o uso de fluidos, há muito menos contrapressão:

insira a descrição da imagem aqui

O Rxjava2 possui algumas estratégias de contrapressão que você pode usar, dependendo do seu caso de uso. Por estratégia, quero dizer que o Rxjava2 fornece uma maneira de lidar com objetos que não podem ser processados ​​devido ao estouro (contrapressão).

aqui estão as estratégias. Não vou passar por todos eles, mas, por exemplo, se você não quiser se preocupar com os itens que estão em excesso, pode usar uma estratégia de queda como esta:

observable.toFlowable (BackpressureStrategy.DROP)

Tanto quanto eu sei, deve haver um limite de 128 itens na fila, depois disso pode haver um estouro (contrapressão). Mesmo que não seja 128, é próximo desse número. Espero que isso ajude alguém.

se você precisar alterar o tamanho do buffer de 128, parece que isso pode ser feito assim (mas observe as restrições de memória:

myObservable.toFlowable(BackpressureStrategy.MISSING).buffer(256); //but using MISSING might be slower.  

no desenvolvimento de software, geralmente a estratégia de contrapressão significa dizer ao emissor que diminua um pouco, pois o consumidor não consegue lidar com a velocidade dos eventos emitidos.

j2emanue
fonte
Eu sempre pensei que contrapressão é o nome de uma família de mecanismos que permitiriam ao consumidor notificar o produtor para diminuir a velocidade ...
kboom
Pode ser o caso. Sim
j2emanue
Existem desvantagens no uso de um Flowable?
IgorGanapolsky
Essas imagens estão mentindo para mim. Eliminar eventos não vai acabar com "mais dinheiro" na parte inferior.
EpicPandaForce 31/03/19
1
@ j2emanue, você está confundindo o tamanho do buffer para os operadores e o operador Flowable.buffer (int). Leia os javadocs com atenção e corrija sua resposta de acordo: reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html
tomek
15

O fato de que seu Flowable travado após emitir 128 valores sem manipulação de contrapressão não significa que sempre travará após exatamente 128 valores: às vezes travará após 10, e às vezes não travará. Acredito que foi isso que aconteceu quando você tentou o exemplo Observable- não houve contrapressão; portanto, seu código funcionou normalmente, da próxima vez que não. A diferença no RxJava 2 é que não há mais conceito de contrapressão em Observables e não há como lidar com isso. Se você estiver projetando uma sequência reativa que provavelmente exigirá manipulação explícita de contrapressão - então Flowableé a sua melhor escolha.

Egor
fonte
Sim, observei que algumas vezes quebrou depois de menos valores, outras vezes não. Mas, novamente, se, por exemplo, eu estou lidando apenas intervalsem backpressure, eu esperaria algum comportamento ou problemas estranhos?
user2141889
Se você tem certeza de que não há como ocorrer problemas de contrapressão em uma sequência observável específica - acho que é bom ignorar a contrapressão.
Egor