RxPY - Hızlı Kılavuz

Bu bölüm reaktif programlamanın ne olduğunu, RxPY'nin ne olduğunu, operatörlerini, özelliklerini, avantajlarını ve dezavantajlarını açıklamaktadır.

Reaktif Programlama nedir?

Reaktif programlama, veri akışı ve değişimin yayılmasıyla ilgilenen bir programlama paradigmasıdır. Bu, bir bileşen tarafından bir veri akışı yayınlandığında, değişikliğin reaktif bir programlama kitaplığı tarafından diğer bileşenlere yayılacağı anlamına gelir. Değişimin yayılması, nihai alıcıya ulaşana kadar devam edecektir.

RxPY'yi kullanarak, eşzamansız veri akışları üzerinde iyi bir kontrole sahip olursunuz, örneğin, URL'ye yapılan bir istek gözlemlenebilir kullanılarak izlenebilir ve yanıt veya hata için istek tamamlandığında dinlemek için gözlemciyi kullanabilirsiniz.

RxPY, asenkron veri akışlarını kullanarak Observables, kullanarak veri akışlarını sorgulayın Operators yani filtre, toplama, birleştirme, eşleme ve aynı zamanda veri akışları için eşzamanlılık kullanma Schedulers. Bir Gözlemlenebilir oluşturma, olması gereken on_next (v), on_error (e) ve on_completed () yöntemleriyle bir gözlemci nesnesi verir.subscribed Böylece bir olay meydana geldiğinde bildirim alırız.

Gözlemlenebilir, boru operatörü kullanılarak bir zincir formatında birden fazla operatör kullanılarak sorgulanabilir.

RxPY, aşağıdakiler gibi çeşitli kategorilerde operatörler sunar: -

• Matematiksel operatörler

• Dönüşüm operatörleri

• Filtreleme operatörleri

• Hata işleme operatörleri

• Kamu hizmeti operatörleri

• Koşullu operatörler

• Oluşturma operatörleri

• Bağlanabilir operatörler

Bu operatörler, bu eğitimde ayrıntılı olarak açıklanmıştır.

RxPy nedir?

RxPY şu şekilde tanımlanır: a library for composing asynchronous and event-based programs using observable collections and pipable query operators in Python RxPy'nin resmi web sitesine göre, https://rxpy.readthedocs.io/en/latest/.

RxPY, Reaktif Programlamayı destekleyen bir python kitaplığıdır. RxPy,Reactive Extensions for Python. Eşzamansız veri aramaları, geri aramalar ve olay tabanlı programlarla ilgilenen reaktif programlama ile çalışmak için gözlemlenebilirleri kullanan bir kitaplıktır.

RxPy'nin Özellikleri

RxPy'de, aşağıdaki kavramlar eşzamansız görevin üstesinden gelir -

Gözlenebilir

Gözlenebilir, bir gözlemci yaratan ve onu, örneğin Tweetler, bilgisayarla ilgili olaylar vb. Gibi beklenen veri akışlarına sahip kaynağa ekleyen bir işlevdir.

Gözlemci

On_next (), on_error () ve on_completed () yöntemlerine sahip bir nesnedir, gözlemlenebilir ile etkileşim olduğunda yani kaynak bir örnek gelen Tweet için etkileşime girdiğinde çağrılır.

Abonelik

Gözlenebilir olan yaratıldığında, gözlemlenebilir olanı yürütmek için ona abone olmamız gerekir.

Operatörler

Operatör, girdi olarak gözlemlenebilir olan saf bir fonksiyondur ve çıktı da gözlemlenebilirdir. Boru operatörünü kullanarak gözlemlenebilir bir veri üzerinde birden fazla operatör kullanabilirsiniz.

Konu

Bir özne, gözlemlenebilir bir dizinin yanı sıra çok noktaya yayın yapabilen bir gözlemcidir, yani abone olan birçok gözlemciyle konuşur. Konu soğuk gözlemlenebilir, yani değerler abone olmuş gözlemciler arasında paylaşılacaktır.

Zamanlayıcılar

RxPy'nin önemli bir özelliği eşzamanlılıktır, yani görevin paralel olarak yürütülmesine izin vermek. Bunu gerçekleştirmek için RxPy'nin zamanlayıcılarla çalışan ve abone olunan görevin yürütülmesine karar verecek olan subscribe_on () ve observe_on () adlı iki operatörü vardır.

RxPY kullanmanın avantajları

Aşağıdakiler RxPy'nin avantajlarıdır -

• Eşzamansız veri akışlarının ve olaylarının işlenmesi söz konusu olduğunda RxPY harika bir kitaplıktır. RxPY, eşzamansız veri aramaları, geri aramalar ve olay tabanlı programlarla ilgilenen reaktif programlama ile çalışmak için gözlemlenebilirleri kullanır.

• RxPY, reaktif programlama ile kullanıldığında hayatı kolaylaştıran matematiksel, dönüşüm, filtreleme, yardımcı program, koşullu, hata işleme, birleştirme kategorilerinde çok sayıda operatör koleksiyonu sunar.

• Eşzamanlılık, yani birden fazla görevin birlikte çalışması, RxPY'deki zamanlayıcılar kullanılarak gerçekleştirilir.

• Eşzamansız görevin yürütülmesi ve paralel işlemin kolaylaştırılması nedeniyle performans RxPY kullanılarak geliştirildi.

RxPY kullanmanın dezavantajı

• Gözlenebilirlerle kodda hata ayıklamak biraz zordur.

Bu bölümde, RxPy'nin kurulumu üzerinde çalışacağız. RxPY ile çalışmaya başlamak için önce Python'u kurmamız gerekiyor. Öyleyse, aşağıdakiler üzerinde çalışacağız -

• Python'u yükleyin
• RxPy'yi yükleyin

Python'u Yükleme

Python resmi sitesine gidin: https://www.python.org/downloads/.aşağıda gösterildiği gibi ve Windows, Linux / Unix ve mac os için mevcut olan en son sürüme tıklayın. Python'u yanınızda bulunan 64 veya 32 bit işletim sisteminize göre indirin.

İndirdikten sonra, .exe file ve sisteminize python kurma adımlarını izleyin.

Python paket yöneticisi, yani pip de yukarıdaki kurulumla birlikte varsayılan olarak kurulacaktır. Sisteminizde global olarak çalışmasını sağlamak için, python'un konumunu PATH değişkenine doğrudan ekleyin, aynı şey kurulumun başlangıcında gösterilir, YOLA EKLE yazan onay kutusunu işaretlemeyi unutmayın. Kontrol etmeyi unutursanız, PATH'e eklemek için lütfen aşağıda verilen adımları izleyin.

PATH'e eklemek için aşağıdaki adımları izleyin -

Bilgisayar simgesine sağ tıklayın ve özellikler → Gelişmiş Sistem Ayarları seçeneğine tıklayın.

Ekranı aşağıda gösterildiği gibi gösterecektir -

Yukarıda gösterildiği gibi Ortam Değişkenlerine tıklayın. Ekranı aşağıda gösterildiği gibi gösterecektir -

Yol'u seçin ve Düzenle düğmesine tıklayın, python'unuzun konum yolunu sonuna ekleyin. Şimdi python sürümünü kontrol edelim.

Python sürümü kontrol ediliyor

E:\pyrx>python --version
Python 3.7.3

RxPY'yi yükleyin

Şimdi, python kurulu olduğuna göre, RxPy'yi kuracağız.

Python kurulduktan sonra, python paket yöneticisi, yani pip de kurulacaktır. Pip sürümünü kontrol etme komutu aşağıdadır -

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

Pip yükledik ve sürümü 19.1.1. Şimdi, RxPy'yi kurmak için pip kullanacağız

Komut aşağıdaki gibidir -

pip install rx

Bu eğitimde, RxPY sürüm 3 ve python sürüm 3.7.3 kullanıyoruz. RxPY sürüm 3'ün çalışması önceki sürümle, yani RxPY sürüm 1 ile biraz farklıdır.

Bu bölümde, 2 sürüm arasındaki farkları ve Python ve RxPY sürümlerini güncellemeniz durumunda yapılması gereken değişiklikleri tartışacağız.

RxPY'de gözlemlenebilir

RxPy sürüm 1'de Observable ayrı bir sınıftı -

from rx import Observable

Gözlemlenebilir'i kullanmak için, onu aşağıdaki gibi kullanmanız gerekir -

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

RxPy sürüm 3'te Gözlemlenebilir, doğrudan rx paketinin bir parçasıdır.

Example

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

RxPy'deki Operatörler

1. sürümde, operatör Gözlemlenebilir sınıfındaki yöntemlerdi. Örneğin, operatörleri kullanmak için Observable'ı aşağıda gösterildiği gibi içe aktarmalıyız -

from rx import Observable

Operatörler, örneğin aşağıda gösterildiği gibi Observable.operatör olarak kullanılır -

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

RxPY sürüm 3 durumunda, operatörler işlev görür ve aşağıdaki şekilde içe aktarılır ve kullanılır -

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

Boru () yöntemini kullanarak Zincirleme Operatörleri

RxPy sürüm 1'de, bir gözlemlenebilir üzerinde birden fazla operatör kullanmanız gerekirse, aşağıdaki gibi yapılmalıdır -

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

Ancak, RxPY sürüm 3 olması durumunda, boru () yöntemini ve aşağıda gösterildiği gibi birden çok operatörü kullanabilirsiniz -

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

Gözlenebilir, bir gözlemci oluşturan ve onu değerlerin beklendiği kaynağa ekleyen bir işlevdir, örneğin tıklamalar, bir dom öğesinden fare olayları vb.

Aşağıda belirtilen konular bu bölümde ayrıntılı olarak incelenecektir.

• Gözlemlenebilirler Oluşturun

• Bir Gözlemlenebilirliğe Abone Olun ve Uygulayın

Gözlenebilirler oluşturun

Bir gözlemlenebilir oluşturmak için kullanacağız create() yöntemi ve aşağıdaki öğelere sahip olan işlevi ona iletin.

• on_next() - Bu işlev, Gözlemlenebilir öğe bir öğe yaydığında çağrılır.

• on_completed() - Gözlemlenebilir tamamlandığında bu işlev çağrılır.

• on_error() - Bu işlev, Gözlemlenebilir'de bir hata oluştuğunda çağrılır.

Create () yöntemiyle çalışmak için önce yöntemi aşağıda gösterildiği gibi içe aktarın -

from rx import create

İşte gözlemlenebilir bir oluşturmak için çalışan bir örnek -

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

Bir Gözlemlenebilirliğe Abone Olun ve Uygulayın

Bir gözlemlenebilirliğe abone olmak için, subscribe () işlevini kullanmamız ve geri arama işlevini on_next, on_error ve on_completed iletmemiz gerekir.

İşte çalışan bir örnek -

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

Subscribe () yöntemi, gözlemlenebilir olanı yürütmeye özen gösterir. Geri arama işlevion_next, on_error ve on_completedabonelik yöntemine geçilmesi gerekir. Abone olma yöntemine çağrı, sırayla test_observable () işlevini yürütür.

Üç geri çağırma işlevinin tümünü subscribe () yöntemine geçirmek zorunlu değildir. Gereksinimlerinize göre on_next (), on_error () ve on_completed () 'i geçebilirsiniz.

Lambda işlevi on_next, on_error ve on_completed için kullanılır. Argümanları alacak ve verilen ifadeyi çalıştıracaktır.

İşte yaratılan gözlemlenebilirin çıktısı -

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

Bu bölümde, RxPY'deki operatörler ayrıntılı olarak açıklanmaktadır. Bu operatörler şunları içerir -

• Operatörlerle Çalışma
• Matematiksel operatörler
• Dönüşüm operatörleri
• Filtreleme operatörleri
• Hata işleme operatörleri
• Kamu hizmeti operatörleri
• Koşullu operatörler
• Oluşturma operatörleri
• Bağlanabilir operatörler
• Operatörleri birleştirmek

Reaktif (Rx) python, python kodlamayla hayatı kolaylaştıran neredeyse birçok operatöre sahiptir. Bu çoklu işleçleri birlikte kullanabilirsiniz, örneğin dizelerle çalışırken harita, filtre, birleştirme işleçlerini kullanabilirsiniz.

Operatörlerle Çalışma

Pipe () yöntemini kullanarak birden çok operatörle birlikte çalışabilirsiniz. Bu yöntem, birden çok işlecin birbirine zincirlenmesine izin verir.

İşte operatörleri kullanmanın çalışan bir örneği -

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

Yukarıdaki örnekte, 1, 2 ve 3 değerlerini alan () yöntemini kullanarak gözlemlenebilir bir yöntem oluşturduk. Şimdi, bu gözlemlenebilir üzerinde, gösterildiği gibi pipe () yöntemini kullanarak herhangi bir sayıda operatörü kullanarak farklı bir işlem gerçekleştirebilirsiniz. yukarıda. Operatörlerin yürütülmesi, verilen gözlemlenebilir üzerinde sırayla devam edecektir.

Operatörlerle çalışmak için önce aşağıda gösterildiği gibi içe aktarın -

from rx import of, operators as op

İşte çalışan bir örnek -

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

Yukarıdaki örnekte, bir filtre operatörü kullanarak çift sayıları filtrelediğimiz ve daha sonra bir azaltma operatörü kullanarak eklediğimiz bir sayı listesi vardır.

Output

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

İşte tartışacağımız Operatörlerin bir listesi -

• Gözlemlenebilirler Yaratmak
• Matematiksel operatörler
• Dönüşüm operatörleri
• Filtreleme operatörleri
• Hata işleme operatörleri
• Kamu hizmeti operatörleri
• Conditional
• Connectable
• Operatörleri birleştirmek

Gözlemlenebilirler Yaratmak

Aşağıda gözlemlenebilirler var, Yaratılış kategorisinde tartışacağız

Örnekleri Göster

Gözlenebilir	Açıklama
oluşturmak	Bu yöntem, bir gözlemlenebilir oluşturmak için kullanılır.
boş	Bu gözlemlenebilir herhangi bir çıktı vermeyecek ve doğrudan tam durumu yaymayacaktır.
asla	Bu yöntem asla tam duruma ulaşamayacak bir gözlemlenebilirlik yaratır.
atmak	Bu yöntem, bir hata atacak bir gözlemlenebilir yaratacaktır.
from_	Bu yöntem, verilen dizi veya nesneyi gözlemlenebilir hale getirecektir.
Aralık	Bu yöntem, bir zaman aşımından sonra üretilen bir dizi değeri verecektir.
sadece	Bu yöntem, verilen değeri gözlemlenebilir hale getirecektir.
Aralık	Bu yöntem, verilen girdiye göre bir tam sayı aralığı verecektir.
tekrar_değer	Bu yöntem, sayıya göre verilen değeri tekrar edecek bir gözlemlenebilir yaratacaktır.
Başlat	Bu yöntem, bir işlevi girdi olarak alır ve girdi işlevinden değer döndüren bir gözlemlenebilir döndürür.
zamanlayıcı	Bu yöntem, zaman aşımı tamamlandıktan sonra değerleri sırayla yayınlayacaktır.

Matematiksel operatörler

Matematiksel operatör kategorisinde tartışacağımız operatörler aşağıdaki gibidir: -

Örnekleri Göster

Şebeke	Açıklama
ortalama	Bu operatör, verilen kaynaktan gözlemlenebilir ortalamayı hesaplayacak ve ortalama değere sahip olacak bir gözlemlenebilir çıktı verecektir.
concat	Bu operatör, iki veya daha fazla gözlemlenebilir nesneyi alacak ve sıradaki tüm değerlere sahip tek bir gözlemlenebilir öğe verecektir.
Miktar	Bu operatör, değerleri olan bir Gözlemlenebilir'i alır ve onu tek bir değere sahip olacak bir Gözlemlenebilir'e dönüştürür. Count işlevi, isteğe bağlı bir bağımsız değişken olarak yüklem işlevini alır. İşlev boolean tipindedir ve yalnızca koşulu sağladığında çıktıya değer katacaktır.
max	Bu operatör, gözlemlenebilir kaynaktan maksimum değeri olan bir gözlemlenebilir verir.
min	Bu operatör, gözlemlenebilir kaynaktan minimum değeri olan bir gözlemlenebilir verecektir.
azaltmak	Bu operatör, gözlemlenebilir kaynaktan gelen değerler üzerinde kullanılan biriktirici işlevi adı verilen bir işlevi alır ve biriken değerleri, biriktirici işlevine geçirilen isteğe bağlı bir çekirdek değerle, gözlemlenebilir biçiminde döndürür.
toplam	Bu operatör, kaynak gözlemlenebilirlerden tüm değerlerin toplamıyla birlikte bir gözlemlenebilir döndürecektir.

Dönüşüm operatörleri

Dönüşüm operatörü kategorisinde tartışacağımız operatörler aşağıda belirtilmiştir -

Örnekleri Göster

Şebeke	Kategori
tampon	Bu operatör, gözlemlenebilir kaynaktan tüm değerleri toplayacak ve verilen sınır koşulu karşılandığında bunları düzenli aralıklarla yayınlayacaktır.
ground_by	Bu operatör, gözlemlenebilir kaynaktan gelen değerleri verilen key_mapper fonksiyonuna göre gruplayacaktır.
harita	Bu operatör, gözlemlenebilir kaynaktan her bir değeri, verilen mapper_func'un çıktısına bağlı olarak yeni bir değere değiştirecektir.
taramak	Bu operatör, gözlemlenebilir kaynaktan gelen değerlere bir toplayıcı işlevi uygulayacak ve yeni değerlerle bir gözlemlenebilir döndürür.

Filtreleme operatörleri

Filtreleme operatörü kategorisinde tartışacağımız operatörler aşağıda verilmiştir -

Örnekleri Göster

Şebeke	Kategori
ifşa etmek	Bu operatör, verilen zaman aralığına kadar gözlemlenebilir kaynaktan değerleri verecek ve geri kalan süreleri göz ardı edecektir.
farklı	Bu operatör, gözlemlenebilir kaynaktan farklı olan tüm değerleri verecektir.
element_at	Bu operatör, verilen indeks için gözlemlenebilir kaynaktan bir eleman verecektir.
filtre	Bu operatör, verilen dayanak fonksiyonuna dayalı olarak gözlemlenebilir kaynaktan değerleri filtreleyecektir.
ilk	Bu operatör, gözlemlenebilir kaynaktan ilk öğeyi verecektir.
ignore_elements	Bu operatör, gözlemlenebilir kaynaktan gelen tüm değerleri göz ardı edecek ve yalnızca geri arama işlevlerini tamamlamak veya hata vermek için çağrıları yürütecektir.
son	Bu operatör, gözlemlenebilir kaynaktan son öğeyi verecektir.
atlama	Bu operatör, girdi olarak alınan sayım öğelerinin ilk oluşumunu atlayacak bir gözlemlenebilir öğeyi geri verecektir.
skip_last	Bu operatör, girdi olarak alınan sayım öğelerinin son oluşumunu atlayacak bir gözlemlenebilir öğeyi geri verecektir.
almak	Bu operatör, verilen sayıya göre sürekli sırayla kaynak değerlerinin bir listesini verecektir.
take_last	Bu operatör, verilen sayıya göre sondan itibaren sürekli sırayla kaynak değerlerinin bir listesini verecektir.

Hata işleme operatörleri

Hata işleme operatörü kategorisinde tartışacağımız operatörler şunlardır: -

Örnekleri Göster

Şebeke	Açıklama
tutmak	Bu operatör, bir istisna olduğunda gözlemlenebilir kaynağı sonlandıracaktır.
yeniden dene	Bu operatör, bir hata olduğunda gözlemlenebilir kaynak üzerinde yeniden deneyecek ve yeniden deneme sayımı tamamlandığında sona erecektir.

Kamu hizmeti operatörleri

Aşağıdakiler, Hizmet operatörü kategorisinde tartışacağımız operatörlerdir.

Örnekleri Göster

Şebeke	Açıklama
gecikme	Bu operatör, kaynak gözlemlenebilir emisyonu verilen saat veya tarihe göre geciktirecektir.
gerçekleştirmek	Bu operatör, gözlemlenebilir kaynaktan değerleri, açık bildirim değerleri biçiminde yayılan değerlerle dönüştürür.
Zaman aralığı	Bu operatör, gözlemlenebilir kaynaktan değerler arasında geçen süreyi verecektir.
zaman aşımı	Bu operatör, geçen süreden sonra gözlemlenebilir kaynaktan tüm değerleri verecek veya aksi takdirde bir hatayı tetikleyecektir.
zaman damgası	Bu operatör, gözlemlenebilir kaynaktan gelen tüm değerlere bir zaman damgası ekleyecektir.

Koşullu ve Boole operatörleri

Koşullu ve Boole Operatör kategorisinde tartışacağımız operatörler aşağıda verildiği gibidir -

Örnekleri Göster

Şebeke	Açıklama
herşey	Bu operatör, gözlemlenebilir kaynaktan gelen tüm değerlerin verilen koşulu karşılayıp karşılamadığını kontrol edecektir.
içerir	Bu operatör, verilen değer mevcutsa ve gözlemlenebilir kaynağın değeriyse, doğru veya yanlış değerine sahip bir gözlemlenebilir döndürür.
default_if_empty	Bu operatör, gözlemlenebilir kaynak boşsa varsayılan bir değer döndürecektir.
sequal_equal	Bu operatör, iki gözlemlenebilir diziyi veya bir dizi değeri karşılaştıracak ve doğru veya yanlış değerine sahip bir gözlemlenebilir döndürür.
skip_until	Bu operatör, ikinci gözlemlenebilir bir değer yayana kadar, gözlemlenebilir kaynaktan değerleri atacaktır.
geç_mek	Bu operatör, geçirilen koşulu karşılayan, gözlemlenebilir kaynaktan değerlere sahip bir gözlemlenebilir döndürecektir.
take_until	Bu operatör, ikinci gözlemlenebilir bir değer yayınladıktan veya sonlandırıldıktan sonra gözlemlenebilir kaynaktan değerleri atacaktır.
take_ while	Bu operatör, koşul başarısız olduğunda gözlemlenebilir kaynaktan değerleri atacaktır.

Bağlanabilir Operatörler

Bağlanabilir Operatör kategorisinde tartışacağımız operatörler şunlardır:

Örnekleri Göster

Şebeke	Açıklama
Yayınla	Bu yöntem, gözlenebilir olanı bağlanabilir bir gözlemlenebilir hale dönüştürecektir.
ref_count	Bu operatör, gözlenebilir olanı normal bir gözlemlenebilir hale getirecektir.
tekrar oynatmak	Bu yöntem replaySubject'e benzer şekilde çalışır. Bu yöntem, gözlemlenebilir zaten yayınlamış ve bazı aboneler abonelikte gecikmiş olsa bile aynı değerleri döndürecektir.

Operatörleri Birleştirme

Aşağıdakiler, Birleştirme operatörü kategorisinde tartışacağımız operatörlerdir.

Örnekleri Göster

Şebeke	Açıklama
Comb_latest	Bu operatör, girdi olarak verilen gözlemlenebilir için bir tuple oluşturacaktır.
birleştirmek	Bu operatör, verilen gözlenebilirleri birleştirecek.
ile başla	Bu operatör verilen değerleri alacak ve gözlemlenebilir kaynağın başlangıcında tam diziye geri dönecektir.
zip	Bu operatör, verilen gözlemlenebilirin ilk değeri alınarak oluşturulan bir demet formundaki değerlere sahip bir gözlemlenebilir döndürür.

Bir özne, gözlemlenebilir bir dizinin yanı sıra, çok noktaya yayın yapabilen bir gözlemcidir, yani abone olan birçok gözlemciyle konuşur.

Konuyla ilgili aşağıdaki konuları tartışacağız -

• Bir konu oluşturun
• Bir konuya abone ol
• Verilerin konuya aktarılması
• BehaviorSubject
• ReplaySubject
• AsyncSubject

Bir konu oluşturun

Bir konuyla çalışmak için, Konuyu aşağıda gösterildiği gibi içe aktarmamız gerekiyor -

from rx.subject import Subject

Aşağıdaki gibi bir konu nesnesi oluşturabilirsiniz -

subject_test = Subject()

Nesne, üç yöntemi olan bir gözlemcidir -

• on_next(value)
• on_error (hata) ve
• on_completed()

Bir Konuya Abone Olun

Aşağıda gösterildiği gibi konu ile ilgili birden fazla abonelik oluşturabilirsiniz -

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

Verileri Konuya Aktarma

Verileri, aşağıda gösterildiği gibi on_next (değer) yöntemini kullanarak oluşturulan konuya aktarabilirsiniz -

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

Veriler konuya eklenen tüm aboneliğe aktarılacaktır.

İşte konunun çalışan bir örneği.

Misal

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

Subject_test nesnesi bir Subject () çağırılarak oluşturulur. Subject_test nesnesinin on_next (değer), on_error (hata) ve on_completed () yöntemlerine başvurusu vardır. Yukarıdaki örneğin çıktısı aşağıda gösterilmiştir -

Çıktı

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

Aşağıda gösterildiği gibi konu yürütmeyi durdurmak için on_completed () yöntemini kullanabiliriz.

Misal

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

Complete'i çağırdığımızda, daha sonra çağrılan bir sonraki yöntem çağrılmaz.

Çıktı

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

Şimdi on_error (hata) yönteminin nasıl çağrılacağını görelim.

Misal

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

Çıktı

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

Davranış Konu

BehaviorSubject, arandığında size en son değeri verecektir. Aşağıda gösterildiği gibi davranış konusu oluşturabilirsiniz -

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

İşte, Davranış Konusu kullanmak için çalışan bir örnek

Misal

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

Çıktı

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

Tekrar Oynatma Konusu

Bir replaysubject, davranış öznesine benzer, burada değerleri tamponlayabilir ve aynısını yeni abonelere yeniden oynatabilir. İşte tekrar konusunun çalışan bir örneği.

Misal

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)

Yeniden oynatma konusunda kullanılan tampon değeri 2'dir. Böylece son iki değer arabelleğe alınacak ve aranan yeni aboneler için kullanılacaktır.

Çıktı

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

AsyncSubject durumunda, çağrılan son değer aboneye iletilir ve yalnızca complete () yöntemi çağrıldıktan sonra yapılır.

Misal

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.

Çıktı

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

RxPy'nin önemli bir özelliği eşzamanlılıktır, yani görevin paralel olarak yürütülmesine izin vermek. Bunu gerçekleştirmek için, abone olunan görevin yürütülmesine karar verecek bir zamanlayıcı ile çalışacak olan subscribe_on () ve observe_on () iki operatörümüz var.

Burada, subscibe_on (), observe_on () ve scheduler'a olan ihtiyacı gösteren çalışan bir örnek var.

Misal

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

Yukarıdaki örnekte 2 görevim var: Görev 1 ve Görev 2. Görevin yürütülmesi sırayla gerçekleşir. İkinci görev yalnızca ilk görev tamamlandığında başlar.

Çıktı

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 birçok Zamanlayıcıyı destekler ve burada, ThreadPoolScheduler'ı kullanacağız. ThreadPoolScheduler, esas olarak mevcut CPU iş parçacıkları ile yönetmeye çalışacaktır.

Örnekte, daha önce gördüğümüz gibi, bize cpu_count verecek bir çoklu işlem modülünden yararlanacağız. Sayım, görevi mevcut iş parçacıkları temelinde paralel olarak çalıştırmayı başaracak olan ThreadPoolScheduler'a verilecektir.

İşte çalışan bir örnek -

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

Yukarıdaki örnekte 2 görevim var ve cpu_count 4'tür. Görev 2 ve bizim için mevcut olan evreler 4 olduğundan, her iki görev de paralel olarak başlayabilir.

Çıktı

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

Çıkışı görürseniz, her iki görev de paralel olarak başlamıştır.

Şimdi, görevin CPU sayısından fazla olduğu, yani CPU sayısının 4 olduğu ve görevlerin 5 olduğu bir senaryo düşünün. Bu durumda, görev tamamlandıktan sonra herhangi bir iş parçacığının boş olup olmadığını kontrol etmemiz gerekir, böylece kuyrukta bulunan yeni göreve atanır.

Bu amaçla, herhangi bir evre boşsa zamanlayıcıyı gözlemleyen observe_on () operatörünü kullanabiliriz. İşte observe_on () kullanan çalışan bir örnek.

Misal

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

Çıktı

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

Çıkışı görürseniz, görev 4 tamamlanır, iş parçacığı bir sonraki göreve verilir, yani görev 5 ve aynısı yürütülmeye başlar.

Bu bölümde aşağıdaki konuları ayrıntılı olarak tartışacağız -

• Gözlemlenebilir, operatörlerin çalışmasını ve gözlemciye abone olmayı gösteren Temel Örnek.
• Gözlemlenebilir ve özne arasındaki fark.
• Soğuk ve sıcak gözlenebilirleri anlamak.

Aşağıda, gözlemlenebilir, operatörlerin ve gözlemciye abone olanların çalışmasını gösteren temel bir örnek verilmiştir.

Misal

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

İşte çok basit bir örnek, burada bu URL'den kullanıcı verilerini alıyorum -

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

Verileri filtrelemek, "C" ile başlayan isimleri vermek ve daha sonra sadece isimleri döndürmek için haritayı kullanmak. İşte aynısı için çıktı -

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

Gözlemlenebilir ve özne arasındaki fark

Bu örnekte, gözlemlenebilir ve özne arasındaki farkı göreceğiz.

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

Çıktı

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

Yukarıdaki örnekte, gözlemlenebilir olana her abone olduğunuzda, size yeni değerler verecektir.

Konu Örneği

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)

Çıktı

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

Değerlerin konuyu kullanan iki abone arasında paylaşıldığını görürseniz.

Soğuk ve Sıcak Gözlemlenebilir Öğeleri Anlamak

Bir gözlemlenebilir şu şekilde sınıflandırılır:

• Soğuk Gözlemlenebilirler
• Sıcak Gözlemlenebilirler

Gözlemlenebilirler arasındaki fark, birden çok abone abone olduğunda fark edilecektir.

Soğuk Gözlemlenebilirler

Soğuk gözlemlenebilirler, yürütülen gözlemlenebilirdir ve her abone olduğunda verileri işler. Abone olunduğunda, gözlemlenebilir çalıştırılır ve taze değerler verilir.

Aşağıdaki örnek, soğuk gözlenebilirlik anlayışını verir.

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

Çıktı

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

Yukarıdaki örnekte, gözlemlenebilir olana her abone olduğunuzda, gözlemlenebilir olanı çalıştıracak ve değerleri yayacaktır. Değerler ayrıca yukarıdaki örnekte gösterildiği gibi aboneden aboneye farklılık gösterebilir.

Sıcak Gözlemlenebilirler

Sıcak gözlemlenebilir olması durumunda, hazır olduklarında değerleri yayarlar ve her zaman bir abonelik için beklemezler. Değerler yayınlandığında tüm aboneler aynı değeri alacaktır.

Gözlenebilir hazır olduğunda değerlerin yayılmasını istediğinizde veya aynı değerleri tüm abonelerinizle paylaşmak istediğinizde sıcak gözlemlenebilirden yararlanabilirsiniz.

Sıcak gözlemlenebilirliğe bir örnek Konu ve bağlanabilir operatörlerdir.

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)

Çıktı

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

Görürseniz aboneler arasında aynı değer paylaşılır. Publish () bağlanabilir gözlemlenebilir işleci kullanarak aynısını elde edebilirsiniz.