RxPY - Làm việc với Chủ đề
Chủ thể là một chuỗi có thể quan sát được, cũng như, một người quan sát có thể phát đa hướng, tức là nói chuyện với nhiều người quan sát đã đăng ký.
Chúng ta sẽ thảo luận các chủ đề sau về chủ đề -
- Tạo một chủ đề
- Đăng ký một chủ đề
- Truyền dữ liệu cho chủ đề
- BehaviorSubject
- ReplaySubject
- AsyncSubject
Tạo một chủ đề
Để làm việc với một chủ đề, chúng ta cần nhập Chủ đề như hình dưới đây -
from rx.subject import Subject
Bạn có thể tạo một chủ thể-đối tượng như sau:
subject_test = Subject()
Đối tượng là một người quan sát có ba phương pháp:
- on_next(value)
- on_error (error) và
- on_completed()
Đăng ký một Chủ đề
Bạn có thể tạo nhiều đăng ký theo chủ đề như hình dưới đây -
subject_test.subscribe(
lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
lambda x: print("The value is {0}".format(x))
)
Truyền dữ liệu cho chủ đề
Bạn có thể chuyển dữ liệu cho chủ đề được tạo bằng phương thức on_next (value) như hình dưới đây -
subject_test.on_next("A")
subject_test.on_next("B")
Dữ liệu sẽ được chuyển cho tất cả các đăng ký, được thêm vào chủ đề.
Đây là một ví dụ hoạt động của chủ đề này.
Thí dụ
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")
Đối tượng subject_test được tạo bằng cách gọi một Subject (). Đối tượng subject_test có tham chiếu đến các phương thức on_next (value), on_error (error) và on_completed (). Kết quả của ví dụ trên được hiển thị bên dưới:
Đầu ra
E:\pyrx>python testrx.py
The value is A
The value is A
The value is B
The value is B
Chúng ta có thể sử dụng phương thức on_completed () để dừng việc thực thi chủ đề như hình dưới đây.
Thí dụ
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")
Khi chúng tôi gọi hoàn tất, phương thức tiếp theo được gọi sau sẽ không được gọi.
Đầu ra
E:\pyrx>python testrx.py
The value is A
The value is A
Bây giờ chúng ta hãy xem, cách gọi phương thức on_error (error).
Thí dụ
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!'))
Đầu ra
E:\pyrx>python testrx.py
Error: There is an Error!
Error: There is an Error!
BehaviorSubject
BehaviorSubject sẽ cung cấp cho bạn giá trị mới nhất khi được gọi. Bạn có thể tạo chủ đề hành vi như hình dưới đây -
from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject"); // initialized the behaviour subject with value:Testing Behaviour Subject
Đây là một ví dụ hoạt động để sử dụng Chủ đề hành vi
Thí dụ
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")
Đầu ra
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
Phát lại chủ đề
Một replayubject tương tự như chủ đề hành vi, trong đó, nó có thể đệm các giá trị và phát lại giống như vậy cho người đăng ký mới. Đây là một ví dụ hoạt động của chủ đề phát lại.
Thí dụ
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)
Giá trị đệm được sử dụng là 2 trên chủ đề phát lại. Vì vậy, hai giá trị cuối cùng sẽ được đệm và sử dụng cho các thuê bao mới được gọi.
Đầu ra
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
Trong trường hợp của AsyncSubject, giá trị cuối cùng được gọi được chuyển cho thuê bao và nó sẽ chỉ được thực hiện sau khi phương thức complete () được gọi.
Thí dụ
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.
Đầu ra
E:\pyrx>python testrx.py
Testing Async Subject A: 2
Testing Async Subject B: 2