RxPY - Szybki przewodnik

W tym rozdziale wyjaśniono, czym jest programowanie reaktywne, czym jest RxPY, jego operatory, cechy, zalety i wady.

Co to jest programowanie reaktywne?

Programowanie reaktywne to paradygmat programowania, który zajmuje się przepływem danych i propagacją zmian. Oznacza to, że gdy przepływ danych jest emitowany przez jeden komponent, zmiana będzie propagowana do innych komponentów przez bibliotekę programowania reaktywnego. Propagacja zmiany będzie kontynuowana, aż dotrze do końcowego odbiorcy.

Korzystając z RxPY, masz dobrą kontrolę nad asynchronicznymi strumieniami danych, na przykład żądanie skierowane do adresu URL może być śledzone za pomocą obserwowalnego i używać obserwatora do nasłuchiwania, kiedy żądanie jest kompletne, w celu uzyskania odpowiedzi lub błędu.

RxPY oferuje obsługę asynchronicznych strumieni danych przy użyciu Observables, odpytuj strumienie danych za pomocą Operators tj. filtruj, sumuj, konkatuj, mapuj, a także wykorzystuj współbieżność dla strumieni danych za pomocą Schedulers. Tworzenie Observable daje obiekt obserwatora z metodami on_next (v), on_error (e) i on_completed (), które muszą byćsubscribed abyśmy otrzymali powiadomienie o wystąpieniu zdarzenia.

Do Observable można zapytać przy użyciu wielu operatorów w formacie łańcucha przy użyciu operatora potoku.

RxPY oferuje operatorów w różnych kategoriach, takich jak: -

• Operatory matematyczne

• Operatory transformacji

• Operatory filtrujące

• Operatory obsługi błędów

• Operatorzy mediów

• Operatory warunkowe

• Operatory tworzenia

• Operatorzy z możliwością podłączenia

Te operatory są szczegółowo wyjaśnione w tym samouczku.

Co to jest RxPy?

RxPY jest zdefiniowany jako a library for composing asynchronous and event-based programs using observable collections and pipable query operators in Python zgodnie z oficjalną stroną RxPy, czyli https://rxpy.readthedocs.io/en/latest/.

RxPY to biblioteka Pythona obsługująca programowanie reaktywne. RxPy oznaczaReactive Extensions for Python. Jest to biblioteka, która używa obserwabli do pracy z programowaniem reaktywnym, które zajmuje się asynchronicznymi wywołaniami danych, wywołaniami zwrotnymi i programami opartymi na zdarzeniach.

Funkcje RxPy

W RxPy następujące koncepcje zajmują się obsługą zadania asynchronicznego -

Zauważalny

Obserwowalna to funkcja, która tworzy obserwatora i dołącza go do źródła zawierającego strumienie danych, których oczekuje się na przykład z tweetów, zdarzeń związanych z komputerem itp.

Obserwator

Jest to obiekt z metodami on_next (), on_error () i on_completed (), który zostanie wywołany, gdy nastąpi interakcja z obserwowalnym, tj. Źródło współdziała z przykładowymi przychodzącymi tweetami itp.

Subskrypcja

Kiedy to, co obserwowalne jest tworzone, aby wykonać to, co obserwowalne, musimy je subskrybować.

Operatorzy

Operator jest czystą funkcją, która przyjmuje obserwowalne jako dane wejściowe, a wyjście jest również obserwowalne. Możesz użyć wielu operatorów na obserwowalnych danych, używając operatora potoku.

Przedmiot

Podmiot jest obserwowalną sekwencją, a także obserwatorem, który może przesyłać multiemisją, tj. Rozmawiać z wieloma obserwatorami, którzy się subskrybowali. Temat jest zimny, obserwowalny, tj. Wartości będą wspólne dla subskrybowanych obserwatorów.

Harmonogramy

Jedną z ważnych cech RxPy jest współbieżność, tj. Umożliwienie równoległego wykonywania zadania. Aby tak się stało, RxPy ma dwa operatory subscribe_on () i observ_on (), które współpracują z programami planującymi i decydują o wykonaniu zasubskrybowanego zadania.

Zalety korzystania z RxPY

Oto zalety RxPy -

• RxPY to niesamowita biblioteka, jeśli chodzi o obsługę strumieni danych i zdarzeń asynchronicznych. RxPY używa obserwabli do pracy z programowaniem reaktywnym, które obsługuje asynchroniczne wywołania danych, wywołania zwrotne i programy oparte na zdarzeniach.

• RxPY oferuje ogromną kolekcję operatorów w matematyce, transformacji, filtrowaniu, użyteczności, warunkowej, obsłudze błędów, łączeniu kategorii, które ułatwiają życie w programowaniu reaktywnym.

• Współbieżność, tj. Jednoczesna praca wielu zadań, jest osiągana za pomocą harmonogramów w RxPY.

• Wydajność jest lepsza przy użyciu RxPY, ponieważ obsługa zadania asynchronicznego i przetwarzanie równoległe są łatwe.

Wada korzystania z RxPY

• Debugowanie kodu za pomocą obserwabli jest trochę trudne.

W tym rozdziale będziemy pracować nad instalacją RxPy. Aby rozpocząć pracę z RxPY, musimy najpierw zainstalować Pythona. Więc będziemy pracować nad następującymi -

• Zainstaluj Python
• Zainstaluj RxPy

Instalowanie Pythona

Przejdź do oficjalnej strony Pythona: https://www.python.org/downloads/.jak pokazano poniżej i kliknij najnowszą wersję dostępną dla systemów Windows, Linux / Unix i mac os. Pobierz Pythona zgodnie z 64- lub 32-bitowym systemem operacyjnym dostępnym u Ciebie.

Po pobraniu kliknij plik .exe file i postępuj zgodnie z instrukcjami, aby zainstalować Pythona w swoim systemie.

Menedżer pakietów Pythona, tj. Pip, zostanie również domyślnie zainstalowany podczas powyższej instalacji. Aby działał globalnie w systemie, dodaj bezpośrednio lokalizację Pythona do zmiennej PATH, to samo jest pokazane na początku instalacji, aby pamiętać o zaznaczeniu pola wyboru, które mówi DODAJ do ŚCIEŻKI. Jeśli zapomnisz to sprawdzić, wykonaj poniższe kroki, aby dodać do PATH.

Aby dodać do PATH, wykonaj poniższe czynności -

Kliknij prawym przyciskiem myszy ikonę komputera i kliknij właściwości → Zaawansowane ustawienia systemu.

Wyświetli ekran, jak pokazano poniżej -

Kliknij Zmienne środowiskowe, jak pokazano powyżej. Wyświetli ekran, jak pokazano poniżej -

Wybierz ścieżkę i kliknij przycisk Edytuj, dodaj na końcu ścieżkę lokalizacji swojego Pythona. Teraz sprawdźmy wersję Pythona.

Sprawdzanie wersji Pythona

E:\pyrx>python --version
Python 3.7.3

Zainstaluj RxPY

Teraz, gdy mamy już zainstalowany Python, zamierzamy zainstalować RxPy.

Po zainstalowaniu Pythona zostanie również zainstalowany menedżer pakietów python, czyli pip. Poniżej znajduje się polecenie sprawdzenia wersji 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)

Mamy zainstalowany pip i wersja to 19.1.1. Teraz użyjemy pip do zainstalowania RxPy

Polecenie jest następujące -

pip install rx

W tym samouczku używamy RxPY w wersji 3 i języka Python w wersji 3.7.3. Działanie RxPY w wersji 3 różni się nieco od wcześniejszej wersji, czyli RxPY w wersji 1.

W tym rozdziale omówimy różnice między dwiema wersjami i zmiany, które należy wprowadzić w przypadku aktualizacji wersji Python i RxPY.

Zaobserwowane w RxPY

W RxPy w wersji 1 Observable była oddzielną klasą -

from rx import Observable

Aby użyć Observable, musisz użyć go w następujący sposób -

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

W RxPy w wersji 3 Observable jest bezpośrednio częścią pakietu rx.

Example

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

Operatory w RxPy

W wersji 1 operatorem były metody z klasy Observable. Na przykład, aby skorzystać z operatorów, musimy zaimportować Observable, jak pokazano poniżej -

from rx import Observable

Operatory są używane jako Observable.operator, na przykład, jak pokazano poniżej -

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)))

W przypadku RxPY w wersji 3 operatory są funkcjami i są importowane i używane w następujący sposób -

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)))

Operatory łańcuchowe przy użyciu metody Pipe ()

W wersji 1 RxPy, gdybyś musiał użyć wielu operatorów na obserwowalnym, należało to zrobić w następujący sposób -

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)))

Ale w przypadku RxPY w wersji 3 możesz użyć metody pipe () i wielu operatorów, jak pokazano poniżej -

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)))

Obserwowalna to funkcja, która tworzy obserwatora i dołącza go do źródła, w którym oczekiwane są wartości, na przykład kliknięcia, zdarzenia myszy z elementu dom itp.

Tematy wymienione poniżej zostaną szczegółowo omówione w tym rozdziale.

• Utwórz obserwowalne

• Zasubskrybuj i wykonaj obserwowalne

Twórz obserwowalne

Aby stworzyć obserwowalne, użyjemy create() metoda i przekaż do niej funkcję, która zawiera następujące elementy.

• on_next() - Ta funkcja jest wywoływana, gdy Observable emituje element.

• on_completed() - Ta funkcja jest wywoływana po zakończeniu Observable.

• on_error() - Ta funkcja jest wywoływana, gdy wystąpi błąd w Observable.

Aby pracować z metodą create (), najpierw zaimportuj metodę, jak pokazano poniżej -

from rx import create

Oto działający przykład tworzenia obserwowalnych -

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).

Zasubskrybuj i wykonaj obserwowalne

Aby zasubskrybować obserwowalną, musimy użyć funkcji subscribe () i przekazać funkcję zwrotną on_next, on_error i on_completed.

Oto działający przykład -

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!"),
)

Metoda subscribe () dba o wykonanie obserwowalnego. Funkcja oddzwanianiaon_next, on_error i on_completedmusi zostać przekazana do metody subscribe. Z kolei wywołanie metody subscribe powoduje wykonanie funkcji test_observable ().

Przekazywanie wszystkich trzech funkcji zwrotnych do metody subscribe () nie jest obowiązkowe. Możesz przekazać zgodnie ze swoimi wymaganiami on_next (), on_error () i on_completed ().

Funkcja lambda jest używana dla on_next, on_error i on_completed. Przyjmie argumenty i wykona podane wyrażenie.

Oto wynik tego, co można zaobserwować -

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

Ten rozdział zawiera szczegółowe informacje o operatorach w RxPY. Te operatory obejmują -

• Praca z operatorami
• Operatory matematyczne
• Operatory transformacji
• Operatory filtrujące
• Operatory obsługi błędów
• Operatorzy mediów
• Operatory warunkowe
• Operatory tworzenia
• Operatorzy z możliwością podłączenia
• Łączenie operatorów

Python reaktywny (Rx) ma prawie wiele operatorów, które ułatwiają życie z kodowaniem w Pythonie. Możesz używać tych wielu operatorów razem, na przykład podczas pracy z łańcuchami możesz używać operatorów mapowania, filtrowania, scalania.

Praca z operatorami

Możesz pracować z wieloma operatorami razem, używając metody pipe (). Ta metoda umożliwia łączenie wielu operatorów razem.

Oto działający przykład użycia operatorów -

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

W powyższym przykładzie stworzyliśmy obserwowalną metodę of (), która przyjmuje wartości 1, 2 i 3. Teraz, na tym obserwowalnym, możesz wykonać inną operację, używając dowolnej liczby operatorów za pomocą metody pipe (), jak pokazano powyżej. Wykonywanie operatorów będzie się odbywać sekwencyjnie na danym obserwowalnym.

Aby pracować z operatorami, najpierw zaimportuj go, jak pokazano poniżej -

from rx import of, operators as op

Oto działający przykład -

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)))

W powyższym przykładzie znajduje się lista liczb, z której filtrujemy liczby parzyste za pomocą operatora filtru, a później dodajemy za pomocą operatora redukuj.

Output

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

Oto lista operatorów, o których będziemy rozmawiać -

• Tworzenie Observables
• Operatory matematyczne
• Operatory transformacji
• Operatory filtrujące
• Operatory obsługi błędów
• Operatorzy mediów
• Conditional
• Connectable
• Łączenie operatorów

Tworzenie Observables

Poniżej znajdują się obserwowalne, które omówimy w kategorii Kreacja

Pokaż przykłady

Zauważalny	Opis
Stwórz	Ta metoda służy do tworzenia obserwowalnych.
pusty	To obserwowalne nic nie wyjdzie i bezpośrednio wyemituje stan kompletny.
nigdy	Ta metoda tworzy obserwowalne, które nigdy nie osiągną pełnego stanu.
rzucać	Ta metoda stworzy obserwowalny, który spowoduje błąd.
od_	Ta metoda skonwertuje podaną tablicę lub obiekt na obserwowalny.
interwał	Ta metoda zwróci serię wartości wygenerowanych po przekroczeniu limitu czasu.
właśnie	Ta metoda zamieni podaną wartość na obserwowalną.
zasięg	Ta metoda zwróci zakres liczb całkowitych na podstawie podanych danych wejściowych.
powtórz_wartość	Ta metoda stworzy obserwowalny, który powtórzy daną wartość zgodnie z podaną liczbą.
początek	Ta metoda przyjmuje funkcję jako dane wejściowe i zwraca obserwowalną, która zwróci wartość z funkcji wejściowej.
regulator czasowy	Ta metoda wyemituje wartości w kolejności po upływie limitu czasu.

Operatory matematyczne

Operatory, które będziemy omawiać w kategorii Operator matematyczny, są następujące: -

Pokaż przykłady

Operator	Opis
średni	Operator ten obliczy średnią z danego obserwowalnego źródła i wyprowadzi obserwowalne, które będą miały średnią wartość.
concat	Ten operator weźmie dwie lub więcej obserwabli i otrzyma jedną obserwowalną ze wszystkimi wartościami w sekwencji.
liczyć	Ten operator przyjmuje Observable with values ​​i konwertuje go na Observable, który będzie miał jedną wartość. Funkcja count przyjmuje funkcję predykatu jako argument opcjonalny. Funkcja jest typu boolean i doda wartość do wyjścia tylko wtedy, gdy spełnia warunek.
max	Ten operator da obserwowalne z maksymalną wartością z obserwowalnego źródła.
min	Ten operator da obserowalną z minimalną wartością z obserwowalnego źródła.
zmniejszyć	Ten operator przyjmuje funkcję zwaną funkcją akumulatora, która jest używana na wartościach pochodzących z obserwowalnego źródła i zwraca skumulowane wartości w postaci obserwowalnej, z opcjonalną wartością początkową przekazaną do funkcji akumulatora.
suma	Ten operator zwróci obserwowalną z sumą wszystkich wartości z obserwabli źródłowych.

Operatory transformacji

Operatory, które będziemy omawiać w kategorii Operator transformacji, są wymienione poniżej -

Pokaż przykłady

Operator	Kategoria
bufor	Operator ten zbierze wszystkie wartości z obserwowalnego źródła i wyemituje je w regularnych odstępach czasu, gdy dany warunek brzegowy zostanie spełniony.
ground_by	Ten operator grupuje wartości pochodzące z obserwowalnego źródła na podstawie podanej funkcji key_mapper.
mapa	Ten operator zmieni każdą wartość z obserwowalnego źródła na nową wartość na podstawie danych wyjściowych funkcji mapper_func.
skanowanie	Ten operator zastosuje funkcję akumulatora do wartości pochodzących z obserwowalnego źródła i zwróci obserwowalną z nowymi wartościami.

Operatory filtrujące

Operatory, które będziemy omawiać w kategorii Operator filtrujący, są podane poniżej -

Pokaż przykłady

Operator	Kategoria
Obalić	Ten operator będzie podawał wartości z obserwowalnego źródła, aż do podanego okresu czasu i zignoruje resztę czasu.
odrębny	Ten operator poda wszystkie wartości, które są różne od obserwowalnego źródła.
element_at	Ten operator da element ze źródła obserwowalny dla podanego indeksu.
filtr	Ten operator odfiltruje wartości z obserwowalnego źródła na podstawie podanej funkcji predykatu.
pierwszy	Ten operator da pierwszy element z obserwowalnego źródła.
ignore_elements	Ten operator zignoruje wszystkie wartości z obserwowalnego źródła i wykona tylko wywołania funkcji zakończenia lub błędu wywołania zwrotnego.
ostatni, ubiegły, zeszły	Ten operator da ostatni obserwowalny element ze źródła.
pominąć	Ten operator zwróci obserwowalną, która pominie pierwsze wystąpienie elementów licznika przyjętych jako dane wejściowe.
skip_last	Ten operator zwróci obserwowalny, który pominie ostatnie wystąpienie elementów licznika przyjętych jako dane wejściowe.
brać	Ten operator poda listę wartości źródłowych w kolejności ciągłej na podstawie podanej liczby.
take_last	Ten operator poda listę wartości źródłowych w kolejności ciągłej od ostatniej na podstawie podanej liczby.

Operatory obsługi błędów

Operatory, które będziemy omawiać w kategorii Operator obsługi błędów to: -

Pokaż przykłady

Operator	Opis
łapać	Ten operator zakończy obserwowalne źródło, gdy wystąpi wyjątek.
spróbować ponownie	Ten operator ponowi próbę na źródle obserwowalnym, gdy wystąpi błąd, a po zliczeniu ponownych prób zakończy się.

Operatorzy mediów

Poniżej przedstawiono operatory, które omówimy w kategorii Operator Utility.

Pokaż przykłady

Operator	Opis
opóźnienie	Operator ten opóźni obserwowalną emisję ze źródła o podaną godzinę lub datę.
zmaterializować	Operator ten przekształci wartości z obserwowalnego źródła na wartości wyemitowane w postaci wartości jawnego powiadomienia.
Przedział czasowy	Ten operator poda czas, jaki upłynął między wartościami z obserwowalnego źródła.
koniec czasu	Ten operator poda wszystkie wartości ze źródła, które można zaobserwować po upływie czasu, w przeciwnym razie spowoduje błąd.
znak czasu	Ten operator dołączy znacznik czasu do wszystkich wartości z obserwowalnego źródła.

Operatory warunkowe i boolowskie

Operatory, które będziemy omawiać w kategorii Operator warunkowy i logiczny, są takie, jak podano poniżej -

Pokaż przykłady

Operator	Opis
wszystko	Ten operator sprawdzi, czy wszystkie wartości z obserwowalnego źródła spełniają podany warunek.
zawiera	Ten operator zwróci obserwowalne z wartością true lub false, jeśli dana wartość jest obecna i jeśli jest to wartość obserwowalnego źródła.
default_if_empty	Ten operator zwróci wartość domyślną, jeśli obserwowalne źródło jest puste.
sekwencja_równa	Ten operator porówna dwie sekwencje obserwabli lub tablicę wartości i zwróci obserwowalne z wartością prawda lub fałsz.
skip_until	Ten operator będzie odrzucał wartości z obserwowalnego źródła, dopóki druga obserwowalna nie wyemituje wartości.
skip_while	Ten operator zwróci obserwowalną z wartościami z obserwowalnego źródła, które spełnia przesłany warunek.
take_until	Ten operator odrzuci wartości ze źródła obserwowalnego po tym, jak druga obserwowalna wyemituje wartość lub zostanie zakończona.
take_while	Ten operator odrzuci wartości ze źródła obserwowalne, gdy warunek nie powiedzie się.

Operatorzy z możliwością podłączenia

Operatorzy, których będziemy omawiać w kategorii Operator z możliwością podłączenia to -

Pokaż przykłady

Operator	Opis
publikować	Ta metoda przekształci to, co obserwowalne w możliwe do podłączenia.
ref_count	Ten operator sprawi, że obserwowalne stanie się normalne.
powtórna rozgrywka	Ta metoda działa podobnie do replaySubject. Ta metoda zwróci te same wartości, nawet jeśli obserwowalne już wyemitowano i niektórzy subskrybenci spóźniają się z subskrypcją.

Łączenie operatorów

Poniżej przedstawiono operatory, które omówimy w kategorii Operator łączący.

Pokaż przykłady

Operator	Opis
connect_latest	Ten operator utworzy krotkę dla obserwowalnego podanego jako dane wejściowe.
łączyć	Ten operator połączy podane obserwable.
zacząć od	Ten operator przyjmie podane wartości i doda na początku obserwowalne źródło, które zwróci pełną sekwencję.
zamek błyskawiczny	Ten operator zwraca obserwowalną z wartościami w postaci krotki, która jest tworzona przez pobranie pierwszej wartości danej obserwowalnej i tak dalej.

Podmiot jest obserwowalną sekwencją, a także obserwatorem, który może przesyłać strumieniowo, tj. Rozmawiać z wieloma subskrybentami.

Omówimy następujące tematy na temat -

• Utwórz temat
• Zapisz się do tematu
• Przekazywanie danych podmiotowi
• BehaviorSubject
• ReplaySubject
• AsyncSubject

Utwórz temat

Aby pracować z tematem, musimy zaimportować temat, jak pokazano poniżej -

from rx.subject import Subject

Możesz utworzyć podmiot-obiekt w następujący sposób -

subject_test = Subject()

Obiekt jest obserwatorem, który ma trzy metody -

• on_next(value)
• on_error (błąd) i
• on_completed()

Zasubskrybuj temat

Możesz utworzyć wiele subskrypcji na ten temat, jak pokazano poniżej -

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

Przekazywanie danych podmiotowi

Możesz przekazać dane podmiotowi utworzonemu za pomocą metody on_next (wartość), jak pokazano poniżej -

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

Dane zostaną przekazane do wszystkich abonamentów, dodanych w temacie.

Oto roboczy przykład tematu.

Przykład

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")

Obiekt subject_test jest tworzony przez wywołanie metody Subject (). Obiekt subject_test odwołuje się do metod on_next (wartość), on_error (błąd) i on_completed (). Dane wyjściowe powyższego przykładu pokazano poniżej -

Wynik

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

Możemy użyć metody on_completed (), aby zatrzymać wykonywanie podmiotu, jak pokazano poniżej.

Przykład

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")

Po wywołaniu complete następna metoda wywoływana później nie jest wywoływana.

Wynik

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

Zobaczmy teraz, jak wywołać metodę on_error (error).

Przykład

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!'))

Wynik

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

BehaviorSubject

BehaviorSubject poda najnowszą wartość po wywołaniu. Możesz stworzyć temat zachowania, jak pokazano poniżej -

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

Oto działający przykład użycia tematu zachowania

Przykład

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")

Wynik

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

Powtórz temat

Obiekt powtórki jest podobny do tematu zachowania, w którym może buforować wartości i odtwarzać je nowym subskrybentom. Oto działający przykład tematu powtórki.

Przykład

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)

Użyta wartość bufora dla tematu powtórki to 2. Zatem ostatnie dwie wartości zostaną zbuforowane i użyte dla nowych wywoływanych abonentów.

Wynik

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

W przypadku AsyncSubject ostatnia wywołana wartość jest przekazywana subskrybentowi i zostanie to zrobione dopiero po wywołaniu metody complete ().

Przykład

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.

Wynik

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

Jedną z ważnych cech RxPy jest współbieżność, tj. Umożliwienie równoległego wykonywania zadania. Aby tak się stało, mamy dwa operatory subscribe_on () i observ_on (), które będą współpracować z harmonogramem, który zadecyduje o wykonaniu zasubskrybowanego zadania.

Oto działający przykład, który pokazuje potrzebę subscibe_on (), obserwa_on () i harmonogramu.

Przykład

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")

W powyższym przykładzie mam 2 zadania: Zadanie 1 i Zadanie 2. Wykonywanie zadania przebiega w kolejności. Drugie zadanie rozpoczyna się dopiero po wykonaniu pierwszego zadania.

Wynik

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 obsługuje wiele programów planujących i tutaj będziemy korzystać z ThreadPoolScheduler. ThreadPoolScheduler będzie głównie próbował zarządzać dostępnymi wątkami procesora.

W przykładzie, który widzieliśmy wcześniej, użyjemy modułu wieloprocesorowego, który da nam cpu_count. Liczba zostanie przekazana ThreadPoolScheduler, który będzie w stanie uruchomić zadanie równolegle na podstawie dostępnych wątków.

Oto działający przykład -

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")

W powyższym przykładzie mam 2 zadania, a cpu_count to 4. Ponieważ zadanie to 2, a dostępnych u nas wątków to 4, oba zadania mogą rozpocząć się równolegle.

Wynik

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

Jeśli widzisz dane wyjściowe, oba zadania zostały uruchomione równolegle.

Rozważmy teraz scenariusz, w którym zadanie jest większe niż liczba procesorów, tj. Liczba procesorów wynosi 4, a zadań 5. W tym przypadku musielibyśmy sprawdzić, czy jakikolwiek wątek został zwolniony po zakończeniu zadania, aby można było przypisane do nowego zadania dostępnego w kolejce.

W tym celu możemy użyć operatora observ_on (), który będzie obserwował planistę, czy jakieś wątki są wolne. Tutaj jest działający przykład z zastosowaniem observ_on ()

Przykład

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")

Wynik

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

Jeśli widzisz dane wyjściowe, w momencie zakończenia zadania 4, wątek jest przekazywany do następnego zadania, tj. Zadania 5 i to samo zaczyna się wykonywać.

W tym rozdziale omówimy szczegółowo następujące tematy -

• Podstawowy przykład pokazujący działanie obserwowalnych, operatorów i subskrybowanie obserwatora.
• Różnica między obserwowalnym a przedmiotem.
• Zrozumienie zimnych i gorących obserwacji.

Poniżej podano podstawowy przykład pokazujący działanie obserwowalnych, operatorów i subskrybowanie obserwatora.

Przykład

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!"),
)

Oto bardzo prosty przykład, w którym pobieram dane użytkownika z tego adresu URL -

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

Filtrowanie danych, aby nadać nazwy zaczynające się od „C”, a później używając mapy, aby zwrócić same nazwy. Oto wynik dla tego samego -

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

Różnica między obserwowalnym a przedmiotem

W tym przykładzie zobaczymy różnicę między obserwowalnym a przedmiotem.

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)))

Wynik

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

W powyższym przykładzie za każdym razem, gdy subskrybujesz to, co obserwowalne, otrzymasz nowe wartości.

Przykład tematu

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)

Wynik

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

Jeśli widzisz, że wartości są wspólne, między obu subskrybentów korzystających z tematu.

Zrozumienie zimnych i gorących obserwacji

Obserwowalny jest klasyfikowany jako

• Zimne obserwowalne
• Hot Observables

Różnica w obserwowalnych zostanie zauważona, gdy subskrybuje wielu abonentów.

Zimne obserwowalne

Zimne obserwowalne są obserwowalne, które są wykonywane i renderuje dane za każdym razem, gdy są subskrybowane. Kiedy jest zasubskrybowany, obserwowalne jest wykonywane i podawane są nowe wartości.

Poniższy przykład daje zrozumienie zimna obserwowalnego.

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)))

Wynik

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

W powyższym przykładzie, za każdym razem, gdy subskrybujesz obserwowalne, wykona obserwowalne i wyemituje wartości. Wartości mogą również różnić się od abonenta do abonenta, jak pokazano w powyższym przykładzie.

Hot Observables

W przypadku obserwowalnych na gorąco, będą emitować wartości, gdy będą gotowe i nie zawsze będą czekać na subskrypcję. Kiedy wartości zostaną wyemitowane, wszyscy abonenci otrzymają tę samą wartość.

Możesz wykorzystać gorące obserwowalne, gdy chcesz, aby wartości były emitowane, gdy obserwowalne są gotowe, lub chcesz udostępnić te same wartości wszystkim swoim subskrybentom.

Przykładem gorących obserwowalnych jest podmiot i operatorzy z możliwością podłączenia.

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)

Wynik

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

Jeśli widzisz, ta sama wartość jest współdzielona między subskrybentami. Możesz osiągnąć to samo za pomocą obserwowalnego operatora publikacji Publish ().