RxPY - Panduan Cepat

Bab ini menjelaskan apa itu pemrograman reaktif, apa itu RxPY, operatornya, fitur, kelebihan dan kekurangannya.

Apa itu Pemrograman Reaktif?

Pemrograman reaktif adalah paradigma pemrograman, yang berhubungan dengan aliran data dan penyebaran perubahan. Artinya, ketika aliran data dipancarkan oleh satu komponen, perubahan tersebut akan disebarkan ke komponen lain oleh pustaka pemrograman reaktif. Penyebaran perubahan akan berlanjut hingga mencapai penerima akhir.

Dengan menggunakan RxPY, Anda memiliki kontrol yang baik pada aliran data asinkron, misalnya, permintaan yang dibuat ke URL dapat dilacak dengan menggunakan observable, dan menggunakan pengamat untuk mendengarkan ketika permintaan selesai untuk respon atau kesalahan.

RxPY menawarkan Anda untuk menangani aliran data asinkron menggunakan Observables, kueri aliran data menggunakan Operators yaitu menyaring, menjumlahkan, menyimpulkan, memetakan dan juga menggunakan konkurensi untuk aliran data menggunakan Schedulers. Membuat Observable, memberikan objek pengamat dengan metode on_next (v), on_error (e) dan on_completed (), yang perlusubscribed agar kita mendapat notifikasi saat suatu peristiwa terjadi.

Observable dapat ditanyakan menggunakan beberapa operator dalam format rantai dengan menggunakan operator pipa.

RxPY menawarkan operator dalam berbagai kategori seperti: -

• Operator matematika

• Operator transformasi

• Operator pemfilteran

• Kesalahan penanganan operator

• Operator utilitas

• Operator bersyarat

• Operator pembuatan

• Operator yang dapat dihubungkan

Operator ini dijelaskan secara rinci dalam tutorial ini.

Apa itu RxPy?

RxPY didefinisikan sebagai a library for composing asynchronous and event-based programs using observable collections and pipable query operators in Python sesuai situs web resmi RxPy, yaitu https://rxpy.readthedocs.io/en/latest/.

RxPY adalah pustaka python untuk mendukung Pemrograman Reaktif. RxPy adalah singkatan dariReactive Extensions for Python. Ini adalah pustaka yang menggunakan observable untuk bekerja dengan pemrograman reaktif yang berhubungan dengan panggilan data asynchronous, callback dan program berbasis event.

Fitur RxPy

Di RxPy, konsep berikut menangani penanganan tugas asinkron -

Tampak

Observable adalah fungsi yang membuat pengamat dan melampirkannya ke sumber yang memiliki aliran data yang diharapkan dari, misalnya, Tweet, peristiwa terkait komputer, dll.

Pengamat

Ini adalah objek dengan metode on_next (), on_error () dan on_completed (), yang akan dipanggil ketika ada interaksi dengan yang dapat diamati, yaitu sumber berinteraksi untuk contoh Tweet masuk, dll.

Berlangganan

Saat observable dibuat, untuk mengeksekusi observable kita perlu berlangganan padanya.

Operator

Operator adalah fungsi murni yang menerima observasi sebagai input dan output juga dapat diamati. Anda dapat menggunakan beberapa operator pada data yang dapat diamati dengan menggunakan operator pipa.

Subyek

Sebuah subjek merupakan urutan yang dapat diamati sekaligus menjadi pengamat yang dapat multicast, yaitu berbicara dengan banyak pengamat yang telah berlangganan. Subjek adalah observable dingin, yaitu nilai-nilai yang akan dibagikan antara pengamat yang telah berlangganan.

Penjadwal

Salah satu fitur penting dari RxPy adalah konkurensi yaitu memungkinkan tugas dijalankan secara paralel. Untuk mewujudkannya RxPy memiliki dua operator subscribe_on () dan observ_on () yang bekerja dengan penjadwal dan akan memutuskan pelaksanaan tugas berlangganan.

Keuntungan menggunakan RxPY

Berikut ini adalah kelebihan RxPy -

• RxPY adalah pustaka yang luar biasa dalam hal penanganan aliran data dan acara asinkron. RxPY menggunakan observable untuk bekerja dengan pemrograman reaktif yang berhubungan dengan panggilan data asynchronous, callback dan program berbasis event.

• RxPY menawarkan banyak koleksi operator dalam matematika, transformasi, pemfilteran, utilitas, bersyarat, penanganan kesalahan, kategori gabungan yang membuat hidup lebih mudah saat digunakan dengan pemrograman reaktif.

• Concurrency yaitu mengerjakan beberapa tugas bersama-sama dicapai dengan menggunakan penjadwal di RxPY.

• Kinerja ditingkatkan menggunakan RxPY karena penanganan tugas asinkron dan pemrosesan paralel menjadi mudah.

Kerugian menggunakan RxPY

• Men-debug kode dengan observable agak sulit.

Pada bab ini, kita akan mengerjakan instalasi RxPy. Untuk mulai bekerja dengan RxPY, kita perlu menginstal Python terlebih dahulu. Jadi, kami akan mengerjakan yang berikut -

• Instal Python
• Instal RxPy

Menginstal Python

Buka situs resmi Python: https://www.python.org/downloads/.seperti yang ditunjukkan di bawah ini, dan klik versi terbaru yang tersedia untuk Windows, Linux / Unix, dan mac os. Unduh Python sesuai OS 64 atau 32-bit Anda yang tersedia untuk Anda.

Setelah Anda mengunduh, klik .exe file dan ikuti langkah-langkah untuk menginstal python di sistem Anda.

Manajer paket python, yaitu pip juga akan diinstal secara default dengan instalasi di atas. Untuk membuatnya berfungsi secara global di sistem Anda, tambahkan langsung lokasi python ke variabel PATH, hal yang sama ditunjukkan di awal instalasi, ingat untuk mencentang kotak, yang bertuliskan TAMBAHKAN ke PATH. Jika, Anda lupa untuk memeriksanya, ikuti langkah-langkah yang diberikan di bawah ini untuk ditambahkan ke PATH.

Untuk menambah PATH ikuti langkah-langkah di bawah ini -

Klik kanan pada ikon Komputer Anda dan klik pada properties → Advanced System Settings.

Ini akan menampilkan layar seperti yang ditunjukkan di bawah ini -

Klik Variabel Lingkungan seperti yang ditunjukkan di atas. Ini akan menampilkan layar seperti yang ditunjukkan di bawah ini -

Pilih Path dan klik tombol Edit, tambahkan jalur lokasi python Anda di bagian akhir. Sekarang, mari kita periksa versi python.

Memeriksa versi python

E:\pyrx>python --version
Python 3.7.3

Instal RxPY

Sekarang, setelah kita menginstal python, kita akan menginstal RxPy.

Setelah python diinstal, pengelola paket python, yaitu pip juga akan diinstal. Berikut adalah perintah untuk memeriksa versi 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)

Kami telah memasang pip dan versinya adalah 19.1.1. Sekarang, kita akan menggunakan pip untuk menginstal RxPy

Perintahnya adalah sebagai berikut -

pip install rx

Dalam tutorial ini, kami menggunakan RxPY versi 3 dan python versi 3.7.3. Cara kerja RxPY versi 3 sedikit berbeda dengan versi sebelumnya, yaitu RxPY versi 1.

Pada bab ini, kita akan membahas perbedaan antara 2 versi dan perubahan yang perlu dilakukan jika Anda memperbarui versi Python dan RxPY.

Dapat diamati di RxPY

Dalam RxPy versi 1, Observable adalah kelas terpisah -

from rx import Observable

Untuk menggunakan Observable, Anda harus menggunakannya sebagai berikut -

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

Dalam RxPy versi 3, Observable secara langsung merupakan bagian dari paket rx.

Example

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

Operator di RxPy

Dalam versi 1, operator adalah metode di kelas Observable. Misalnya, untuk menggunakan operator, kita harus mengimpor Observable seperti yang ditunjukkan di bawah ini -

from rx import Observable

Operator digunakan sebagai Observable.operator, misalnya, seperti yang ditunjukkan di bawah ini -

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

Dalam kasus RxPY versi 3, operator berfungsi dan diimpor dan digunakan sebagai berikut -

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

Operator Rantai Menggunakan metode Pipa ()

Dalam RxPy versi 1, jika Anda harus menggunakan banyak operator pada sebuah observable, itu harus dilakukan sebagai berikut -

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

Tetapi, dalam kasus RxPY versi 3, Anda dapat menggunakan metode pipa () dan beberapa operator seperti yang ditunjukkan di bawah ini -

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

Observable, adalah fungsi yang membuat pengamat dan melampirkannya ke sumber di mana nilai diharapkan, misalnya, klik, peristiwa mouse dari elemen dom, dll.

Topik-topik yang disebutkan di bawah ini akan dipelajari secara rinci dalam bab ini.

• Buat Observable

• Berlangganan dan Jalankan Observable

Buat observable

Untuk membuat sebuah observable, kita akan menggunakan create() metode dan meneruskan fungsi itu yang memiliki item berikut.

• on_next() - Fungsi ini dipanggil saat Observable memancarkan item.

• on_completed() - Fungsi ini dipanggil saat Observable selesai.

• on_error() - Fungsi ini dipanggil ketika terjadi kesalahan pada Observable.

Untuk bekerja dengan metode create () pertama-tama impor metode seperti yang ditunjukkan di bawah ini -

from rx import create

Berikut adalah contoh kerja, untuk membuat sebuah observable -

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

Berlangganan dan Jalankan Observable

Untuk berlangganan observable, kita perlu menggunakan fungsi subscribe () dan meneruskan fungsi callback on_next, on_error dan on_completed.

Berikut adalah contoh yang berfungsi -

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

Metode subscribe () menangani eksekusi yang dapat diamati. Fungsi panggilan balikon_next, on_error dan on_completedharus diteruskan ke metode berlangganan. Metode panggilan untuk berlangganan, pada gilirannya, menjalankan fungsi test_observable ().

Tidak wajib meneruskan ketiga fungsi callback ke metode subscribe (). Anda dapat meneruskan sesuai kebutuhan Anda on_next (), on_error () dan on_completed ().

Fungsi lambda digunakan untuk on_next, on_error dan on_completed. Ini akan mengambil argumen dan mengeksekusi ekspresi yang diberikan.

Berikut adalah output dari observasi yang dibuat -

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

Bab ini menjelaskan tentang operator di RxPY secara detail. Operator ini termasuk -

• Bekerja dengan Operator
• Operator matematika
• Operator transformasi
• Operator pemfilteran
• Kesalahan penanganan operator
• Operator utilitas
• Operator bersyarat
• Operator pembuatan
• Operator yang dapat dihubungkan
• Menggabungkan operator

Python reaktif (Rx) memiliki hampir banyak operator, yang memudahkan hidup dengan coding python. Anda dapat menggunakan beberapa operator ini bersama-sama, misalnya, saat bekerja dengan string Anda dapat menggunakan operator peta, filter, gabungkan.

Bekerja dengan Operator

Anda bisa bekerja dengan banyak operator bersama-sama menggunakan metode pipa (). Metode ini memungkinkan merangkai beberapa operator secara bersamaan.

Di sini, adalah contoh kerja menggunakan operator -

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

Dalam contoh di atas, kami telah membuat observable menggunakan metode () yang mengambil nilai 1, 2 dan 3. Sekarang, pada observasi ini, Anda dapat melakukan operasi yang berbeda, menggunakan sejumlah operator menggunakan metode pipa () seperti yang ditunjukkan atas. Eksekusi operator akan berlanjut secara berurutan pada yang dapat diamati.

Untuk bekerja dengan operator, impor dulu seperti yang ditunjukkan di bawah ini -

from rx import of, operators as op

Di sini, adalah contoh yang berfungsi -

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

Dalam contoh di atas, ada daftar angka, dari mana kami memfilter angka genap menggunakan operator filter dan kemudian menambahkannya menggunakan operator pengurangan.

Output

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

Berikut adalah daftar Operator, yang akan kita bahas -

• Membuat Observable
• Operator matematika
• Operator transformasi
• Operator pemfilteran
• Kesalahan penanganan operator
• Operator utilitas
• Conditional
• Connectable
• Menggabungkan operator

Membuat Observable

Berikut adalah observasi yang akan kita bahas dalam kategori Penciptaan

Tunjukkan Contoh

Tampak	Deskripsi
membuat	Metode ini digunakan untuk membuat sebuah observable.
kosong	Observable ini tidak akan menampilkan apa pun dan langsung memancarkan status lengkap.
tidak pernah	Metode ini membuat observasi yang tidak akan pernah mencapai status lengkap.
melemparkan	Metode ini akan membuat sebuah observable yang akan memunculkan error.
dari_	Metode ini akan mengubah larik atau objek yang diberikan menjadi yang dapat diamati.
selang	Metode ini akan memberikan serangkaian nilai yang dihasilkan setelah batas waktu.
hanya	Metode ini akan mengubah nilai yang diberikan menjadi yang dapat diamati.
jarak	Metode ini akan memberikan kisaran bilangan bulat berdasarkan masukan yang diberikan.
nilai_ulang	Metode ini akan membuat sebuah observable yang akan mengulang nilai yang diberikan sesuai hitungan yang diberikan.
Mulailah	Metode ini mengambil fungsi sebagai masukan dan mengembalikan yang dapat diamati yang akan mengembalikan nilai dari fungsi masukan.
timer	Metode ini akan mengeluarkan nilai secara berurutan setelah batas waktu selesai.

Operator matematika

Operator yang akan kita bahas dalam kategori Operator matematika adalah sebagai berikut: -

Tunjukkan Contoh

Operator	Deskripsi
rata-rata	Operator ini akan menghitung rata-rata dari sumber yang dapat diamati yang diberikan dan mengeluarkan yang dapat diamati yang akan memiliki nilai rata-rata.
concat	Operator ini akan mengambil dua atau lebih observable dan diberi satu observable dengan semua nilai dalam urutan.
menghitung	Operator ini mengambil Observable dengan nilai dan mengubahnya menjadi Observable yang akan memiliki satu nilai. Fungsi count mengambil fungsi predikat sebagai argumen opsional. Fungsinya berjenis boolean dan akan menambah nilai ke keluaran hanya jika memenuhi syarat.
maks	Operator ini akan memberikan observable dengan nilai maksimal dari source observable.
min	Operator ini akan memberikan nilai yang dapat diamati dengan nilai min dari sumber yang dapat diamati.
mengurangi	Operator ini mengambil fungsi yang disebut fungsi akumulator yang digunakan pada nilai yang berasal dari sumber yang dapat diamati, dan mengembalikan nilai yang terakumulasi dalam bentuk yang dapat diamati, dengan nilai benih opsional yang diteruskan ke fungsi akumulator.
jumlah	Operator ini akan mengembalikan yang dapat diamati dengan jumlah semua nilai dari sumber yang dapat diamati.

Operator transformasi

Operator yang akan kita diskusikan dalam kategori operator Transformasi disebutkan di bawah -

Tunjukkan Contoh

Operator	Kategori
penyangga	Operator ini akan mengumpulkan semua nilai dari sumber yang dapat diamati, dan memancarkannya secara berkala setelah kondisi batas yang diberikan terpenuhi.
ground_by	Operator ini akan mengelompokkan nilai yang berasal dari sumber yang dapat diamati berdasarkan fungsi key_mapper yang diberikan.
peta	Operator ini akan mengubah setiap nilai dari sumber yang dapat diamati menjadi nilai baru berdasarkan keluaran dari mapper_func yang diberikan.
memindai	Operator ini akan menerapkan fungsi akumulator ke nilai yang berasal dari sumber yang bisa diamati dan mengembalikan yang bisa diamati dengan nilai baru.

Operator pemfilteran

Operator yang akan kita diskusikan dalam kategori Operator pemfilteran diberikan di bawah ini -

Tunjukkan Contoh

Operator	Kategori
debounce	Operator ini akan memberikan nilai dari sumber yang dapat diamati, hingga rentang waktu yang diberikan dan mengabaikan sisa waktu yang telah berlalu.
berbeda	Operator ini akan memberikan semua nilai yang berbeda dari sumber yang dapat diamati.
element_at	Operator ini akan memberikan elemen dari sumber yang dapat diamati untuk indeks yang diberikan.
Saring	Operator ini akan memfilter nilai dari sumber yang dapat diamati berdasarkan fungsi predikat yang diberikan.
pertama	Operator ini akan memberikan elemen pertama dari sumber yang dapat diamati.
ignore_elements	Operator ini akan mengabaikan semua nilai dari sumber yang dapat diamati dan hanya menjalankan panggilan ke fungsi callback lengkap atau kesalahan.
terakhir	Operator ini akan memberikan elemen terakhir dari sumber yang dapat diamati.
melewatkan	Operator ini akan mengembalikan sebuah observasi yang akan melewatkan kejadian pertama dari item hitungan yang diambil sebagai input.
skip_last	Operator ini akan mengembalikan sebuah observasi yang akan melewatkan kejadian terakhir dari item penghitungan yang diambil sebagai input.
mengambil	Operator ini akan memberikan daftar nilai sumber secara berurutan berdasarkan jumlah yang diberikan.
take_last	Operator ini akan memberikan daftar nilai sumber secara berurutan dari yang terakhir berdasarkan hitungan yang diberikan.

Kesalahan penanganan operator

Operator yang akan kita diskusikan dalam kategori Operator penanganan kesalahan adalah: -

Tunjukkan Contoh

Operator	Deskripsi
menangkap	Operator ini akan menghentikan sumber yang dapat diamati jika ada pengecualian.
mencoba kembali	Operator ini akan mencoba kembali pada sumber yang dapat diamati ketika ada kesalahan dan setelah penghitungan percobaan ulang selesai, itu akan dihentikan.

Operator utilitas

Berikut ini adalah operator yang akan kita diskusikan dalam kategori operator Utilitas.

Tunjukkan Contoh

Operator	Deskripsi
menunda	Operator ini akan menunda emisi yang dapat diamati sumber sesuai waktu atau tanggal yang diberikan.
mewujudkan	Operator ini akan mengonversi nilai dari sumber yang dapat diamati dengan nilai yang dipancarkan dalam bentuk nilai pemberitahuan eksplisit.
jarak waktu	Operator ini akan memberikan waktu yang berlalu antara nilai-nilai dari sumber yang dapat diamati.
waktu habis	Operator ini akan memberikan semua nilai dari sumber yang dapat diamati setelah waktu yang berlalu atau akan memicu kesalahan.
cap waktu	Operator ini akan memasang stempel waktu ke semua nilai dari sumber yang dapat diamati.

Operator bersyarat dan Boolean

Operator yang akan kita diskusikan dalam kategori Operator Bersyarat dan Boolean adalah seperti yang diberikan di bawah ini -

Tunjukkan Contoh

Operator	Deskripsi
semua	Operator ini akan memeriksa apakah semua nilai dari sumber yang dapat diamati memenuhi kondisi yang diberikan.
mengandung	Operator ini akan mengembalikan yang dapat diamati dengan nilai benar atau salah jika nilai yang diberikan ada dan jika itu adalah nilai sumber yang dapat diamati.
default_if_empty	Operator ini akan mengembalikan nilai default jika observasi sumber kosong.
sequence_equal	Operator ini akan membandingkan dua urutan yang dapat diamati atau larik nilai dan mengembalikan yang dapat diamati dengan nilai benar atau salah.
skip_until	Operator ini akan membuang nilai dari sumber yang dapat diamati hingga yang dapat diamati kedua memancarkan nilai.
lewati_waktu	Operator ini akan mengembalikan sebuah observable dengan nilai-nilai dari source observable yang memenuhi kondisi yang dilalui.
take_until	Operator ini akan membuang nilai dari sumber yang dapat diamati setelah pengamatan kedua memancarkan nilai atau dihentikan.
take_time	Operator ini akan membuang nilai dari sumber yang dapat diamati ketika kondisi gagal.

Operator Terhubung

Operator yang akan kita diskusikan dalam kategori Connectable Operator adalah -

Tunjukkan Contoh

Operator	Deskripsi
menerbitkan	Metode ini akan mengubah observable menjadi observable yang dapat dihubungkan.
ref_count	Operator ini akan membuat observable menjadi observable normal.
ulangan	Cara kerja metode ini mirip dengan replaySubject. Metode ini akan mengembalikan nilai yang sama, bahkan jika observasi telah dipancarkan dan beberapa pelanggan terlambat berlangganan.

Menggabungkan Operator

Berikut ini adalah operator yang akan kita diskusikan dalam kategori operator Menggabungkan.

Tunjukkan Contoh

Operator	Deskripsi
menggabungkan_latest	Operator ini akan membuat tupel untuk observasi yang diberikan sebagai input.
menggabungkan	Operator ini akan menggabungkan observable yang diberikan.
dimulai dari	Operator ini akan mengambil nilai yang diberikan dan menambahkan di awal sumber yang dapat diamati kembali ke urutan penuh.
zip	Operator ini mengembalikan yang dapat diamati dengan nilai dalam bentuk tupel yang dibentuk dengan mengambil nilai pertama dari pengamatan yang diberikan dan seterusnya.

Subjek adalah urutan yang dapat diamati, serta, pengamat yang dapat multicast, yaitu berbicara dengan banyak pengamat yang telah berlangganan.

Kami akan membahas topik berikut tentang subjek -

• Buat subjek
• Berlangganan ke subjek
• Meneruskan data ke subjek
• BehaviorSubject
• ReplaySubject
• AsyncSubject

Buat subjek

Untuk bekerja dengan subjek, kita perlu mengimpor Subjek seperti yang ditunjukkan di bawah ini -

from rx.subject import Subject

Anda dapat membuat subjek-objek sebagai berikut -

subject_test = Subject()

Objek adalah pengamat yang memiliki tiga metode -

• on_next(value)
• on_error (kesalahan) dan
• on_completed()

Berlangganan ke Subjek

Anda dapat membuat beberapa langganan pada subjek seperti yang ditunjukkan di bawah ini -

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

Meneruskan Data ke Subjek

Anda dapat mengirimkan data ke subjek yang dibuat menggunakan metode on_next (nilai) seperti yang ditunjukkan di bawah ini -

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

Data akan diteruskan ke semua langganan, ditambahkan pada subjek.

Di sini, adalah contoh kerja dari subjek tersebut.

Contoh

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

Objek subject_test dibuat dengan memanggil Subject (). Objek subject_test memiliki referensi ke metode on_next (nilai), on_error (error) dan on_completed (). Output dari contoh di atas ditunjukkan di bawah ini -

Keluaran

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

Kita bisa menggunakan metode on_completed (), untuk menghentikan eksekusi subjek seperti yang ditunjukkan di bawah ini.

Contoh

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

Setelah kami memanggil selesai, metode selanjutnya yang dipanggil nanti tidak dipanggil.

Keluaran

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

Sekarang mari kita lihat, bagaimana memanggil metode on_error (error).

Contoh

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

Keluaran

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

BehaviorSubject

BehaviorSubject akan memberi Anda nilai terbaru saat dipanggil. Anda dapat membuat subjek perilaku seperti yang ditunjukkan di bawah ini -

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

Di sini, adalah contoh kerja untuk menggunakan Behavior Subject

Contoh

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

Keluaran

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

Putar Ulang Subjek

Sebuah replaysubject mirip dengan subjek perilaku, di mana, ia dapat menyangga nilai-nilai dan memutar ulang hal yang sama ke pelanggan baru. Di sini, adalah contoh kerja subjek ulangan.

Contoh

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)

Nilai buffer yang digunakan adalah 2 pada subjek ulangan. Jadi, dua nilai terakhir akan di-buffer dan digunakan untuk pelanggan baru yang dipanggil.

Keluaran

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

Dalam kasus AsyncSubject, nilai terakhir yang dipanggil diteruskan ke pelanggan, dan itu akan dilakukan hanya setelah metode complete () dipanggil.

Contoh

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.

Keluaran

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

Salah satu fitur penting dari RxPy adalah konkurensi, yaitu memungkinkan tugas dijalankan secara paralel. Untuk mewujudkannya, kami memiliki dua operator subscribe_on () dan observ_on () yang akan bekerja dengan penjadwal, yang akan memutuskan pelaksanaan tugas berlangganan.

Di sini, adalah contoh kerja, yang menunjukkan kebutuhan untuk subscibe_on (), observ_on () dan penjadwal.

Contoh

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

Dalam contoh di atas, saya memiliki 2 tugas: Tugas 1 dan Tugas 2. Pelaksanaan tugas secara berurutan. Tugas kedua hanya dimulai, saat tugas pertama selesai.

Keluaran

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 mendukung banyak Scheduler, dan di sini, kita akan menggunakan ThreadPoolScheduler. ThreadPoolScheduler terutama akan mencoba mengelola dengan utas CPU yang tersedia.

Dalam contoh, kita telah melihat sebelumnya, kita akan menggunakan modul multiprosesing yang akan memberi kita cpu_count. Hitungannya akan diberikan ke ThreadPoolScheduler yang akan mengatur agar tugas bekerja secara paralel berdasarkan utas yang tersedia.

Di sini, adalah contoh yang berfungsi -

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

Dalam contoh di atas, saya memiliki 2 tugas dan cpu_count adalah 4. Karena, tugasnya adalah 2 dan utas yang tersedia bersama kami adalah 4, keduanya tugas dapat dimulai secara paralel.

Keluaran

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

Jika Anda melihat hasilnya, kedua tugas telah dimulai secara paralel.

Sekarang, pertimbangkan skenario, di mana tugas lebih dari jumlah CPU yaitu jumlah CPU 4 dan tugas 5. Dalam kasus ini, kita perlu memeriksa apakah ada utas yang bebas setelah tugas selesai, sehingga, itu bisa ditugaskan ke tugas baru yang tersedia dalam antrian.

Untuk tujuan ini, kita dapat menggunakan operator observ_on () yang akan mengamati penjadwal jika ada utas yang bebas. Di sini, adalah contoh kerja menggunakan observ_on ()

Contoh

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

Keluaran

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

Jika Anda melihat hasilnya, saat tugas 4 selesai, utas diberikan ke tugas berikutnya yaitu, tugas 5 dan tugas yang sama mulai dijalankan.

Dalam bab ini, kita akan membahas topik-topik berikut secara rinci -

• Contoh Dasar yang menunjukkan cara kerja observable, operator, dan berlangganan pengamat.
• Perbedaan antara observasi dan subjek.
• Memahami pengamatan dingin dan panas.

Diberikan di bawah ini adalah contoh dasar yang menunjukkan cara kerja observable, operator, dan berlangganan pengamat.

Contoh

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

Ini, adalah contoh yang sangat sederhana, di mana, saya mendapatkan data pengguna dari URL ini -

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

Memfilter data, untuk memberi nama yang dimulai dengan "C", dan kemudian menggunakan peta untuk mengembalikan nama saja. Ini adalah hasil yang sama -

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

Perbedaan antara observasi dan subjek

Dalam contoh ini, kita akan melihat perbedaan antara yang dapat diamati dan subjek.

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

Keluaran

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

Dalam contoh di atas, setiap kali Anda berlangganan observable, itu akan memberi Anda nilai baru.

Contoh Subjek

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)

Keluaran

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

Jika Anda melihat nilai-nilai dibagikan, antara kedua pelanggan menggunakan subjek.

Memahami Observable Dingin dan Panas

Sebuah observasi diklasifikasikan sebagai

• Observable Dingin
• Observable Panas

Perbedaan yang dapat diamati akan terlihat saat beberapa pelanggan berlangganan.

Observable Dingin

Dapat diamati dingin, dapat diamati yang dijalankan, dan membuat data setiap kali itu berlangganan. Ketika berlangganan, observasi dijalankan dan nilai segar diberikan.

Contoh berikut memberikan pemahaman tentang observasi dingin.

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

Keluaran

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

Dalam contoh di atas, setiap kali Anda berlangganan observable, itu akan mengeksekusi nilai observable dan emit. Nilainya juga dapat berbeda dari pelanggan ke pelanggan seperti yang ditunjukkan pada contoh di atas.

Observable Panas

Dalam kasus hot observable, mereka akan memancarkan nilai saat siap dan tidak akan selalu menunggu langganan. Ketika nilai-nilai itu dipancarkan, semua pelanggan akan mendapatkan nilai yang sama.

Anda dapat menggunakan hot observable saat Anda ingin nilai dipancarkan saat observable sudah siap, atau Anda ingin membagikan nilai yang sama ke semua pelanggan Anda.

Contoh pengamatan panas adalah Subjek dan operator yang dapat dihubungkan.

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)

Keluaran

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

Jika Anda melihat, nilai yang sama dibagikan di antara pelanggan. Anda dapat mencapai hal yang sama menggunakan operator observasi yang dapat dihubungkan dengan publish ().