RxPY - Guia Rápido

Este capítulo explica o que é programação reativa, o que é RxPY, seus operadores, recursos, vantagens e desvantagens.

O que é programação reativa?

A programação reativa é um paradigma de programação que lida com o fluxo de dados e a propagação da mudança. Isso significa que, quando um fluxo de dados é emitido por um componente, a mudança será propagada para outros componentes por uma biblioteca de programação reativa. A propagação da mudança continuará até atingir o receptor final.

Ao usar o RxPY, você tem um bom controle sobre os fluxos de dados assíncronos, por exemplo, uma solicitação feita para URL pode ser rastreada usando observável e usar o observador para ouvir quando a solicitação for concluída para resposta ou erro.

RxPY oferece a você lidar com fluxos de dados assíncronos usando Observables, consulte os fluxos de dados usando Operators ou seja, filtrar, somar, concatenar, mapear e também usar a simultaneidade para os fluxos de dados usando Schedulers. Criando um Observable, dá um objeto observador com os métodos on_next (v), on_error (e) e on_completed (), que precisa sersubscribed para que recebamos uma notificação quando um evento ocorrer.

O Observable pode ser consultado usando vários operadores em um formato de cadeia usando o operador de tubo.

RxPY oferece operadoras em várias categorias como: -

  • Operadores matemáticos

  • Operadores de transformação

  • Operadores de filtragem

  • Operadores de tratamento de erros

  • Operadores de serviços públicos

  • Operadores condicionais

  • Operadores de criação

  • Operadores conectáveis

Esses operadores são explicados em detalhes neste tutorial.

O que é RxPy?

RxPY é definido como a library for composing asynchronous and event-based programs using observable collections and pipable query operators in Python conforme o site oficial da RxPy, que é https://rxpy.readthedocs.io/en/latest/.

RxPY é uma biblioteca python para dar suporte à programação reativa. RxPy significaReactive Extensions for Python. É uma biblioteca que usa observáveis ​​para trabalhar com programação reativa que lida com chamadas de dados assíncronas, retornos de chamada e programas baseados em eventos.

Recursos do RxPy

No RxPy, os conceitos a seguir cuidam de lidar com a tarefa assíncrona -

Observável

Um observável é uma função que cria um observador e o anexa à fonte com fluxos de dados esperados de, por exemplo, Tweets, eventos relacionados ao computador, etc.

Observador

É um objeto com os métodos on_next (), on_error () e on_completed (), que será chamado quando houver interação com o observável, isto é, a fonte interage para um exemplo de Tweets recebidos, etc.

Inscrição

Quando o observável é criado, para executar o observável, precisamos assiná-lo.

Operadores

Um operador é uma função pura que recebe observáveis ​​como entrada e a saída também é observável. Você pode usar vários operadores em dados observáveis ​​usando o operador pipe.

Sujeito

Um assunto é uma sequência observável, bem como um observador que pode fazer multicast, ou seja, falar com muitos observadores que se inscreveram. O assunto é um observável frio, ou seja, os valores serão compartilhados entre os observadores que foram inscritos.

Schedulers

Uma característica importante do RxPy é a simultaneidade, ou seja, permitir que a tarefa seja executada em paralelo. Para que isso aconteça, o RxPy tem dois operadores subscribe_on () e observe_on () que trabalham com agendadores e decidem a execução da tarefa inscrita.

Vantagens de usar RxPY

A seguir estão as vantagens do RxPy -

  • RxPY é uma biblioteca incrível quando se trata de lidar com fluxos de dados assíncronos e eventos. RxPY usa observáveis ​​para trabalhar com programação reativa que lida com chamadas de dados assíncronas, retornos de chamada e programas baseados em eventos.

  • O RxPY oferece uma enorme coleção de operadores em categorias matemáticas, de transformação, de filtragem, de utilidade, condicionais, de tratamento de erros e de junção que tornam a vida mais fácil quando usados ​​com programação reativa.

  • A simultaneidade, isto é, trabalhar em várias tarefas em conjunto, é obtida usando agendadores no RxPY.

  • O desempenho é aprimorado usando RxPY, pois o manuseio de tarefas assíncronas e o processamento paralelo são facilitados.

Desvantagem de usar RxPY

  • Depurar o código com observáveis ​​é um pouco difícil.

Neste capítulo, trabalharemos na instalação do RxPy. Para começar a trabalhar com RxPY, precisamos primeiro instalar o Python. Então, vamos trabalhar no seguinte -

  • Instale Python
  • Instale RxPy

Instalando Python

Vá para o site oficial do Python: https://www.python.org/downloads/.conforme mostrado abaixo, e clique na versão mais recente disponível para Windows, Linux / Unix e mac os. Baixe Python de acordo com seu sistema operacional de 64 ou 32 bits disponível com você.

Depois de fazer o download, clique no .exe file e siga as etapas para instalar o python em seu sistema.

O gerenciador de pacotes python, ou seja, pip também será instalado por padrão com a instalação acima. Para fazer funcionar globalmente em seu sistema, adicione diretamente a localização do python à variável PATH, a mesma é mostrada no início da instalação, para lembrar de marcar a caixa de seleção, que diz ADD to PATH. Caso você se esqueça de verificá-lo, siga os passos abaixo indicados para adicionar ao PATH.

Para adicionar ao PATH, siga as etapas abaixo -

Clique com o botão direito no ícone do seu computador e clique em propriedades → Configurações avançadas do sistema.

Ele exibirá a tela conforme mostrado abaixo -

Clique em Variáveis ​​de ambiente como mostrado acima. Ele exibirá a tela conforme mostrado abaixo -

Selecione o caminho e clique no botão Editar, adicione o caminho da localização do seu python no final. Agora, vamos verificar a versão do python.

Verificando a versão do python

E:\pyrx>python --version
Python 3.7.3

Instale RxPY

Agora que temos o python instalado, vamos instalar o RxPy.

Assim que o python for instalado, o gerenciador de pacotes python, ou seja, o pip também será instalado. A seguir está o comando para verificar a versão do pip -

E:\pyrx>pip --version
pip 19.1.1 from c:\users\xxxx\appdata\local\programs\python\python37\lib\site-
packages\pip (python 3.7)

Temos o pip instalado e a versão é 19.1.1. Agora, vamos usar o pip para instalar o RxPy

O comando é o seguinte -

pip install rx

Neste tutorial, estamos usando RxPY versão 3 e python versão 3.7.3. O funcionamento do RxPY versão 3 difere um pouco da versão anterior, ou seja, o RxPY versão 1.

Neste capítulo, vamos discutir as diferenças entre as 2 versões e as mudanças que precisam ser feitas no caso de você estar atualizando as versões Python e RxPY.

Observável em RxPY

Na versão 1 do RxPy, Observable era uma classe separada -

from rx import Observable

Para usar o Observable, você deve usá-lo da seguinte maneira -

Observable.of(1,2,3,4,5,6,7,8,9,10)

Na versão 3 do RxPy, o Observable faz parte diretamente do pacote rx.

Example

import rx
rx.of(1,2,3,4,5,6,7,8,9,10)

Operadores em RxPy

Na versão 1, o operador era métodos na classe Observable. Por exemplo, para fazer uso de operadores, temos que importar Observable conforme mostrado abaixo -

from rx import Observable

Os operadores são usados ​​como Observable.operator, por exemplo, conforme mostrado abaixo -

Observable.of(1,2,3,4,5,6,7,8,9,10)\
   .filter(lambda i: i %2 == 0) \
   .sum() \
   .subscribe(lambda x: print("Value is {0}".format(x)))

No caso do RxPY versão 3, os operadores funcionam, são importados e usados ​​da seguinte forma -

import rx
from rx import operators as ops
rx.of(1,2,3,4,5,6,7,8,9,10).pipe(
   ops.filter(lambda i: i %2 == 0),
   ops.sum()
).subscribe(lambda x: print("Value is {0}".format(x)))

Operadores de encadeamento usando o método Pipe ()

Na versão 1 do RxPy, caso você tivesse que usar vários operadores em um observável, isso tinha que ser feito da seguinte maneira -

Example

from rx import Observable
Observable.of(1,2,3,4,5,6,7,8,9,10)\
   .filter(lambda i: i %2 == 0) \
   .sum() \
   .subscribe(lambda x: print("Value is {0}".format(x)))

Mas, no caso do RxPY versão 3, você pode usar o método pipe () e vários operadores como mostrado abaixo -

Example

import rx
from rx import operators as ops
rx.of(1,2,3,4,5,6,7,8,9,10).pipe(
   ops.filter(lambda i: i %2 == 0),
   ops.sum()
).subscribe(lambda x: print("Value is {0}".format(x)))

Um observável é uma função que cria um observador e o anexa à fonte onde os valores são esperados, por exemplo, cliques, eventos de mouse de um elemento dom, etc.

Os tópicos mencionados abaixo serão estudados detalhadamente neste capítulo.

  • Criar observáveis

  • Assine e execute um observável

Crie observáveis

Para criar um observável, usaremos create() método e passar a função para ele que contém os seguintes itens.

  • on_next() - Esta função é chamada quando o Observable emite um item.

  • on_completed() - Esta função é chamada quando o Observable é concluído.

  • on_error() - Esta função é chamada quando ocorre um erro no Observable.

Para trabalhar com o método create (), primeiro importe o método conforme mostrado abaixo -

from rx import create

Aqui está um exemplo prático, para criar um observável -

testrx.py

from rx import create
deftest_observable(observer, scheduler):
   observer.on_next("Hello")
   observer.on_error("Error")
   observer.on_completed()
source = create(test_observable).

Assine e execute um observável

Para assinar um observável, precisamos usar a função subscribe () e passar a função de retorno de chamada on_next, on_error e on_completed.

Aqui está um exemplo prático -

testrx.py

from rx import create
deftest_observable(observer, scheduler):
   observer.on_next("Hello")
   observer.on_completed()
source = create(test_observable)
source.subscribe(
   on_next = lambda i: print("Got - {0}".format(i)),
   on_error = lambda e: print("Error : {0}".format(e)),
   on_completed = lambda: print("Job Done!"),
)

O método subscribe () cuida da execução do observável. A função de retorno de chamadaon_next, on_error e on_completeddeve ser passado para o método de inscrição. A chamada para o método de inscrição, por sua vez, executa a função test_observable ().

Não é obrigatório passar todas as três funções de retorno de chamada para o método subscribe (). Você pode passar de acordo com seus requisitos para on_next (), on_error () e on_completed ().

A função lambda é usada para on_next, on_error e on_completed. Ele pegará os argumentos e executará a expressão fornecida.

Aqui está a saída do observável criado -

E:\pyrx>python testrx.py
Got - Hello
Job Done!

Este capítulo explica sobre os operadores em RxPY em detalhes. Esses operadores incluem -

  • Trabalhando com Operadores
  • Operadores matemáticos
  • Operadores de transformação
  • Operadores de filtragem
  • Operadores de tratamento de erros
  • Operadores de serviços públicos
  • Operadores condicionais
  • Operadores de criação
  • Operadores conectáveis
  • Combinando operadores

O python reativo (Rx) tem quase muitos operadores, que facilitam a vida com a codificação python. Você pode usar esses vários operadores juntos, por exemplo, ao trabalhar com strings, você pode usar os operadores de mapa, filtro e mesclagem.

Trabalhando com Operadores

Você pode trabalhar com vários operadores juntos usando o método pipe (). Este método permite encadear vários operadores juntos.

Aqui está um exemplo prático do uso de operadores -

test = of(1,2,3) // an observable
subscriber = test.pipe(
   op1(),
   op2(),
   op3()
)

No exemplo acima, criamos um método observável of () que leva os valores 1, 2 e 3. Agora, neste observável, você pode realizar uma operação diferente, usando qualquer número de operadores usando o método pipe (), conforme mostrado acima. A execução dos operadores continuará sequencialmente no observável dado.

Para trabalhar com operadores, primeiro importe-o conforme mostrado abaixo -

from rx import of, operators as op

Aqui está um exemplo prático -

testrx.py

from rx import of, operators as op
test = of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
sub1 = test.pipe(
   op.filter(lambda s: s%2==0),
   op.reduce(lambda acc, x: acc + x)
)
sub1.subscribe(lambda x: print("Sum of Even numbers is {0}".format(x)))

No exemplo acima, há uma lista de números, da qual estamos filtrando números pares usando um operador de filtro e, posteriormente, adicionando-o usando um operador de redução.

Output

E:\pyrx>python testrx.py
Sum of Even numbers is 30

Aqui está uma lista de operadores, que vamos discutir -

  • Criação de observáveis
  • Operadores matemáticos
  • Operadores de transformação
  • Operadores de filtragem
  • Operadores de tratamento de erros
  • Operadores de serviços públicos
  • Conditional
  • Connectable
  • Combinando operadores

Criação de observáveis

A seguir estão os observáveis, vamos discutir na categoria Criação

Mostrar exemplos

Observável Descrição
crio Este método é usado para criar um observável.
vazio Este observável não produzirá nada e emitirá diretamente o estado completo.
Nunca Este método cria um observável que nunca alcançará o estado completo.
lançar Este método criará um observável que gerará um erro.
de_ Este método converterá a matriz ou objeto fornecido em um observável.
intervalo Este método fornecerá uma série de valores produzidos após um tempo limite.
somente Este método converterá determinado valor em um observável.
alcance Este método fornecerá um intervalo de inteiros com base na entrada fornecida.
repeat_value Este método criará um observável que repetirá o valor fornecido de acordo com a contagem fornecida.
começar Este método recebe uma função como entrada e retorna um valor observável que retornará o valor da função de entrada.
cronômetro Este método emitirá os valores em sequência após o tempo limite terminar.

Operadores matemáticos

Os operadores que vamos discutir na categoria de operadores matemáticos são os seguintes: -

Mostrar exemplos

Operador Descrição
média Este operador irá calcular a média a partir da fonte observável fornecida e produzir um observável que terá o valor médio.
concat Esse operador pegará dois ou mais observáveis ​​e receberá um único observável com todos os valores na sequência.
contagem

Este operador obtém um Observable com valores e o converte em um Observable que terá um único valor. A função de contagem assume a função de predicado como um argumento opcional.

A função é do tipo booleano e adicionará valor à saída somente se atender à condição.
max Este operador fornecerá um observável com valor máximo da fonte observável.
min Este operador fornecerá um valor observável com valor mínimo da fonte observável.
reduzir Este operador assume uma função chamada função acumuladora que é usada nos valores vindos da fonte observável e retorna os valores acumulados na forma de um observável, com um valor semente opcional passado para a função acumuladora.
soma Este operador retornará um observável com a soma de todos os valores dos observáveis ​​de origem.

Operadores de transformação

Os operadores que vamos discutir na categoria Operador de transformação são mencionados abaixo -

Mostrar exemplos

Operador Categoria
amortecedor Esse operador coletará todos os valores da fonte observável e os emitirá em intervalos regulares assim que a condição de limite fornecida for satisfeita.
ground_by Este operador agrupará os valores provenientes da fonte observável com base na função key_mapper fornecida.
mapa Este operador irá alterar cada valor da fonte observável em um novo valor com base na saída do mapper_func fornecido.
Varredura Este operador aplicará uma função de acumulador aos valores provenientes da fonte observável e retornará um observável com novos valores.

Operadores de filtragem

Os operadores que discutiremos na categoria de operador de filtragem são fornecidos abaixo -

Mostrar exemplos

Operador Categoria
debounce Este operador fornecerá os valores da fonte observável, até que o intervalo de tempo fornecido e ignore o restante do tempo.
distinto Este operador fornecerá todos os valores que são distintos da fonte observável.
element_at Este operador fornecerá um elemento da fonte observável para o índice fornecido.
filtro Este operador filtrará os valores da fonte observável com base na função de predicado fornecida.
primeiro Este operador fornecerá o primeiro elemento da fonte observável.
ignore_elements Este operador irá ignorar todos os valores observáveis ​​de origem e apenas executará chamadas para funções de callback completas ou com erro.
último Este operador fornecerá o último elemento da fonte observável.
pular Este operador retornará um observável que ignorará a primeira ocorrência de itens de contagem tomados como entrada.
skip_last Este operador retornará um observável que irá ignorar a última ocorrência de itens de contagem tomados como entrada.
levar Este operador fornecerá uma lista de valores de origem em ordem contínua com base na contagem fornecida.
take_last Este operador fornecerá uma lista de valores de origem em ordem contínua a partir do último com base na contagem fornecida.

Operadores de tratamento de erros

Os operadores que vamos discutir na categoria Operador de tratamento de erros são: -

Mostrar exemplos

Operador Descrição
pegar Este operador encerrará a fonte observável quando houver uma exceção.
tentar novamente Este operador tentará novamente na fonte observável quando houver um erro e, assim que a contagem de novas tentativas for concluída, ela será encerrada.

Operadores de serviços públicos

A seguir estão os operadores que discutiremos na categoria Operador de serviços públicos.

Mostrar exemplos

Operador Descrição
demora Este operador irá atrasar a emissão observável da fonte de acordo com a hora ou data fornecida.
materializar Este operador converterá os valores da fonte observável com os valores emitidos na forma de valores de notificação explícitos.
intervalo de tempo Este operador dará o tempo decorrido entre os valores da fonte observável.
tempo esgotado Este operador fornecerá todos os valores da fonte observáveis ​​após o tempo decorrido ou irá disparar um erro.
timestamp Este operador anexará um carimbo de data / hora a todos os valores do observável de origem.

Operadores condicionais e booleanos

Os operadores que vamos discutir na categoria Operador condicional e booleano são os indicados abaixo -

Mostrar exemplos

Operador Descrição
todos Este operador verificará se todos os valores da fonte observável satisfazem a condição fornecida.
contém Este operador retornará um observável com o valor verdadeiro ou falso se o valor fornecido estiver presente e se for o valor da fonte observável.
default_if_empty Este operador retornará um valor padrão se a fonte observável estiver vazia.
sequência_equal Este operador irá comparar duas sequências de observáveis ​​ou uma matriz de valores e retornar um observável com o valor verdadeiro ou falso.
skip_until Este operador descartará os valores da fonte observável até que o segundo observável emita um valor.
skip_while Este operador retornará um observável com valores do observável de origem que satisfaça a condição passada.
pegue_até Este operador descartará os valores da fonte observável após o segundo observável emitir um valor ou ser encerrado.
take_while Este operador descartará os valores da fonte observáveis ​​quando a condição falhar.

Operadores conectáveis

Os operadores que vamos discutir na categoria Operador conectável são -

Mostrar exemplos

Operador Descrição
publicar Este método converterá o observável em um observável conectável.
ref_count Este operador tornará o observável um observável normal.
repetir Este método funciona de forma semelhante ao replaySubject. Este método retornará os mesmos valores, mesmo se o observável já tiver sido emitido e alguns dos assinantes estiverem atrasados ​​na assinatura.

Operadores de combinação

A seguir estão os operadores que discutiremos na categoria de operador de combinação.

Mostrar exemplos

Operador Descrição
combinar o último Este operador criará uma tupla para o observável dado como entrada.
fundir Este operador irá mesclar dados observáveis.
começar com Este operador pegará os valores fornecidos e adicionará no início do retorno observável de origem a sequência completa.
fecho eclair Este operador retorna um observável com valores em uma forma de tupla, que é formado tomando o primeiro valor do observável fornecido e assim por diante.

Um assunto é uma sequência observável, bem como um observador que pode fazer multicast, ou seja, falar com muitos observadores que se inscreveram.

Vamos discutir os seguintes tópicos sobre o assunto -

  • Crie um assunto
  • Inscreva-se em um assunto
  • Passando dados para o assunto
  • BehaviorSubject
  • ReplaySubject
  • AsyncSubject

Crie um assunto

Para trabalhar com um assunto, precisamos importar Assunto conforme mostrado abaixo -

from rx.subject import Subject

Você pode criar um objeto-sujeito da seguinte maneira -

subject_test = Subject()

O objeto é um observador que possui três métodos -

  • on_next(value)
  • on_error (erro) e
  • on_completed()

Inscrever-se em um assunto

Você pode criar várias assinaturas sobre o assunto, conforme mostrado abaixo -

subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)

Passando Dados para o Assunto

Você pode passar dados para o assunto criado usando o método on_next (valor) como mostrado abaixo -

subject_test.on_next("A")
subject_test.on_next("B")

Os dados serão repassados ​​a todas as inscrições, agregadas no assunto.

Aqui está um exemplo prático do assunto.

Exemplo

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_next("B")

O objeto subject_test é criado chamando um Subject (). O objeto subject_test faz referência aos métodos on_next (value), on_error (error) e on_completed (). O resultado do exemplo acima é mostrado abaixo -

Resultado

E:\pyrx>python testrx.py
The value is A
The value is A
The value is B
The value is B

Podemos usar o método on_completed (), para parar a execução do assunto como mostrado abaixo.

Exemplo

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_completed()
subject_test.on_next("B")

Assim que chamarmos complete, o próximo método chamado posteriormente não será invocado.

Resultado

E:\pyrx>python testrx.py
The value is A
The value is A

Vamos agora ver como chamar o método on_error (error).

Exemplo

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.on_error(Exception('There is an Error!'))

Resultado

E:\pyrx>python testrx.py
Error: There is an Error!
Error: There is an Error!

BehaviorSubject

BehaviorSubject fornecerá o valor mais recente quando chamado. Você pode criar um assunto de comportamento conforme mostrado abaixo -

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject"); // initialized the behaviour subject with value:Testing Behaviour Subject

Aqui está um exemplo prático para usar o assunto de comportamento

Exemplo

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject");
behavior_subject.subscribe(
   lambda x: print("Observer A : {0}".format(x))
)
behavior_subject.on_next("Hello")
behavior_subject.subscribe(
   lambda x: print("Observer B : {0}".format(x))
)
behavior_subject.on_next("Last call to Behaviour Subject")

Resultado

E:\pyrx>python testrx.py
Observer A : Testing Behaviour Subject
Observer A : Hello
Observer B : Hello
Observer A : Last call to Behaviour Subject
Observer B : Last call to Behaviour Subject

Assunto Repetir

Um sujeito de replay é semelhante ao sujeito de comportamento, em que ele pode armazenar os valores e reproduzir os mesmos para os novos assinantes. Aqui está um exemplo prático do assunto de repetição.

Exemplo

from rx.subject import ReplaySubject
replay_subject = ReplaySubject(2)
replay_subject.subscribe(lambda x: print("Testing Replay Subject A: {0}".format(x)))
replay_subject.on_next(1)
replay_subject.on_next(2)
replay_subject.on_next(3)
replay_subject.subscribe(lambda x: print("Testing Replay Subject B: {0}".format(x)));
replay_subject.on_next(5)

O valor do buffer usado é 2 no assunto de reprodução. Portanto, os dois últimos valores serão armazenados em buffer e usados ​​para os novos assinantes chamados.

Resultado

E:\pyrx>python testrx.py
Testing Replay Subject A: 1
Testing Replay Subject A: 2
Testing Replay Subject A: 3
Testing Replay Subject B: 2
Testing Replay Subject B: 3
Testing Replay Subject A: 5
Testing Replay Subject B: 5

AsyncSubject

No caso de AsyncSubject, o último valor chamado é passado para o assinante, e isso será feito somente depois que o método complete () for chamado.

Exemplo

from rx.subject import AsyncSubject
async_subject = AsyncSubject()
async_subject.subscribe(lambda x: print("Testing Async Subject A: {0}".format(x)))
async_subject.on_next(1)
async_subject.on_next(2)
async_subject.on_completed()
async_subject.subscribe(lambda x: print("Testing Async Subject B: {0}".format(x)))
Here, before complete is called, the last value passed to the subject is 2, and the same is given to the subscribers.

Resultado

E:\pyrx>python testrx.py
Testing Async Subject A: 2
Testing Async Subject B: 2

Uma característica importante do RxPy é a simultaneidade, ou seja, permitir que a tarefa seja executada em paralelo. Para que isso aconteça, temos dois operadores subscribe_on () e observe_on () que trabalharão com um planejador, que decidirá a execução da tarefa inscrita.

Aqui, está um exemplo de trabalho, que mostra a necessidade de subscibe_on (), observe_on () e planejador.

Exemplo

import random
import time
import rx
from rx import operators as ops
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
) 
input("Press any key to exit\n")

No exemplo acima, tenho 2 tarefas: Tarefa 1 e Tarefa 2. A execução da tarefa está em sequência. A segunda tarefa começa apenas, quando a primeira tarefa é concluída.

Resultado

E:\pyrx>python testrx.py
From Task 1: 1
From Task 1: 2
From Task 1: 3
From Task 1: 4
From Task 1: 5
Task 1 complete
From Task 2: 1
From Task 2: 2
From Task 2: 3
From Task 2: 4
Task 2 complete

RxPy suporta muitos Scheduler, e aqui, vamos fazer uso de ThreadPoolScheduler. ThreadPoolScheduler tentará principalmente gerenciar com os threads de CPU disponíveis.

No exemplo que vimos anteriormente, vamos fazer uso de um módulo de multiprocessamento que nos dará o cpu_count. A contagem será dada ao ThreadPoolScheduler que conseguirá fazer a tarefa funcionar em paralelo com base nos threads disponíveis.

Aqui está um exemplo prático -

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
input("Press any key to exit\n")

No exemplo acima, eu tenho 2 tarefas e cpu_count é 4. Como a tarefa é 2 e os threads disponíveis conosco são 4, ambas as tarefas podem iniciar em paralelo.

Resultado

E:\pyrx>python testrx.py
Cpu count is : 4
Press any key to exit
From Task 1: 1
From Task 2: 1
From Task 1: 2
From Task 2: 2
From Task 2: 3
From Task 1: 3
From Task 2: 4
Task 2 complete
From Task 1: 4
From Task 1: 5
Task 1 complete

Se você vir a saída, ambas as tarefas foram iniciadas em paralelo.

Agora, considere um cenário, onde a tarefa é maior do que a contagem de CPU, ou seja, a contagem de CPU é 4 e as tarefas são 5. Nesse caso, precisaríamos verificar se algum encadeamento ficou livre após a conclusão da tarefa, para que possa ser atribuído à nova tarefa disponível na fila.

Para este propósito, podemos usar o operador observe_on () que observará o escalonador se algum encadeamento estiver livre. Aqui está um exemplo de trabalho usando observe_on ()

Exemplo

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
#Task 3
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 3: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 3 complete")
)
#Task 4
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 4: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 4 complete")
)
#Task 5
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.observe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 5: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 5 complete")
)
input("Press any key to exit\n")

Resultado

E:\pyrx>python testrx.py
Cpu count is : 4
From Task 4: 1
From Task 4: 2
From Task 1: 1
From Task 2: 1
From Task 3: 1
From Task 1: 2
From Task 3: 2
From Task 4: 3
From Task 3: 3
From Task 2: 2
From Task 1: 3
From Task 4: 4
Task 4 complete
From Task 5: 1
From Task 5: 2
From Task 5: 3
From Task 3: 4
Task 3 complete
From Task 2: 3
Press any key to exit
From Task 5: 4
Task 5 complete
From Task 1: 4
From Task 2: 4
Task 2 complete
From Task 1: 5
Task 1 complete

Se você ver a saída, no momento em que a tarefa 4 é concluída, o thread é passado para a próxima tarefa, ou seja, a tarefa 5 e a mesma começa a ser executada.

Neste capítulo, discutiremos os seguintes tópicos em detalhes -

  • Exemplo básico mostrando o funcionamento do observável, operadores e assinando o observador.
  • Diferença entre observável e sujeito.
  • Compreendendo os observáveis ​​quentes e frios.

A seguir, é fornecido um exemplo básico que mostra o funcionamento do observável, dos operadores e da assinatura do observador.

Exemplo

test.py

import requests
import rx
import json
from rx import operators as ops
def filternames(x):
   if (x["name"].startswith("C")):
      return x["name"]
   else :
      return ""
content = requests.get('https://jsonplaceholder.typicode.com/users')
y = json.loads(content.text)
source = rx.from_(y)
case1 = source.pipe(
   ops.filter(lambda c: filternames(c)),
   ops.map(lambda a:a["name"])
)
case1.subscribe(
   on_next = lambda i: print("Got - {0}".format(i)), 8. RxPy — Examples
   on_error = lambda e: print("Error : {0}".format(e)),
   on_completed = lambda: print("Job Done!"),
)

Aqui está um exemplo muito simples, em que estou obtendo dados do usuário a partir deste URL -

https://jsonplaceholder.typicode.com/users.

Filtrando os dados, para dar os nomes começando com "C", e depois usando o mapa para retornar apenas os nomes. Aqui está a saída para o mesmo -

E:\pyrx\examples>python test.py
Got - Clementine Bauch
Got - Chelsey Dietrich
Got - Clementina DuBuque
Job Done!

Diferença entre observável e sujeito

Neste exemplo, veremos a diferença entre um observável e um sujeito.

from rx import of, operators as op
import random
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
print("From first subscriber")
subscriber1 = sub1.subscribe(lambda i: print("From sub1 {0}".format(i)))
print("From second subscriber")
subscriber2 = sub1.subscribe(lambda i: print("From sub2 {0}".format(i)))

Resultado

E:\pyrx>python testrx.py
From first subscriber
From sub1 1.610450821095726
From sub1 2.9567564032037335
From sub1 3.933217537811936
From sub1 4.82444905626622
From sub1 5.929414892567188
From second subscriber
From sub2 1.8573813517529874
From sub2 2.902433239469483
From sub2 3.2289868093016825
From sub2 4.050413890694411
From sub2 5.226515068012821

No exemplo acima, cada vez que você assina o observável, ele fornecerá novos valores.

Exemplo de Assunto

from rx import of, operators as op
import random
from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("From sub1 {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("From sub2 {0}".format(x))
)
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
subscriber = sub1.subscribe(subject_test)

Resultado

E:\pyrx>python testrx.py
From sub1 1.1789422863284509
From sub2 1.1789422863284509
From sub1 2.5525627903260153
From sub2 2.5525627903260153
From sub1 3.4191549324778325
From sub2 3.4191549324778325
From sub1 4.644042420199624
From sub2 4.644042420199624
From sub1 5.079896897489065
From sub2 5.079896897489065

Se você ver que os valores são compartilhados, entre os dois assinantes usando o assunto.

Compreendendo os observáveis ​​quentes e frios

Um observável é classificado como

  • Observáveis ​​Frios
  • Observáveis ​​quentes

A diferença nos observáveis ​​será notada quando vários assinantes estiverem se inscrevendo.

Observáveis ​​Frios

Observáveis ​​frios, são observáveis ​​que são executados e renderizam dados cada vez que são inscritos. Quando é inscrito, o observável é executado e os novos valores são fornecidos.

O exemplo a seguir fornece a compreensão do frio observável.

from rx import of, operators as op
import random
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
print("From first subscriber")
subscriber1 = sub1.subscribe(lambda i: print("From sub1 {0}".format(i)))
print("From second subscriber")
subscriber2 = sub1.subscribe(lambda i: print("From sub2 {0}".format(i)))

Resultado

E:\pyrx>python testrx.py
From first subscriber
From sub1 1.610450821095726
From sub1 2.9567564032037335
From sub1 3.933217537811936
From sub1 4.82444905626622
From sub1 5.929414892567188
From second subscriber
From sub2 1.8573813517529874
From sub2 2.902433239469483
From sub2 3.2289868093016825
From sub2 4.050413890694411
From sub2 5.226515068012821

No exemplo acima, toda vez que você assinar o observável, ele executará o observável e emitirá valores. Os valores também podem diferir de assinante para assinante, conforme mostrado no exemplo acima.

Observáveis ​​quentes

No caso de hot observable, eles emitirão os valores quando estiverem prontos e nem sempre aguardarão por uma assinatura. Quando os valores são emitidos, todos os assinantes receberão o mesmo valor.

Você pode fazer uso do observável quente quando quiser que os valores sejam emitidos quando o observável estiver pronto, ou quiser compartilhar os mesmos valores com todos os seus assinantes.

Um exemplo de observável quente são os operadores sujeitos e conectáveis.

from rx import of, operators as op
import random
from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("From sub1 {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("From sub2 {0}".format(x))
)
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
subscriber = sub1.subscribe(subject_test)

Resultado

E:\pyrx>python testrx.py
From sub1 1.1789422863284509
From sub2 1.1789422863284509
From sub1 2.5525627903260153
From sub2 2.5525627903260153
From sub1 3.4191549324778325
From sub2 3.4191549324778325
From sub1 4.644042420199624
From sub2 4.644042420199624
From sub1 5.079896897489065
From sub2 5.079896897489065

Se você vir, o mesmo valor é compartilhado entre os assinantes. Você pode obter o mesmo usando o operador observável conectável publish ().