RxJava - Hướng dẫn nhanh

RxJava là một phần mở rộng dựa trên Java của ReactiveX. Nó cung cấp triển khai hoặc dự án ReactiveX trong Java. Sau đây là các đặc điểm chính của RxJava.

• Mở rộng mẫu người quan sát.

• Hỗ trợ chuỗi dữ liệu / sự kiện.

• Cung cấp cho các toán tử soạn các chuỗi với nhau một cách khai báo.

• Xử lý các cấu trúc dữ liệu phân luồng, đồng bộ, an toàn và đồng thời trong nội bộ.

ReactiveX là gì?

ReactiveX là một dự án nhằm cung cấp khái niệm lập trình phản ứng cho các ngôn ngữ lập trình khác nhau. Lập trình phản ứng đề cập đến kịch bản trong đó chương trình phản ứng khi và khi dữ liệu xuất hiện. Nó là một khái niệm lập trình dựa trên sự kiện và các sự kiện có thể truyền đến các quan sát viên đăng ký.

Theo Reactive, họ đã kết hợp những gì tốt nhất của mẫu Observer, mẫu Iterator và mẫu chức năng.

Mẫu Observer được thực hiện đúng. ReactiveX là sự kết hợp của những ý tưởng tốt nhất từ ​​mẫu Observer, mẫu Iterator và lập trình chức năng.

Lập trình chức năng

Lập trình chức năng xoay quanh việc xây dựng phần mềm sử dụng các chức năng thuần túy. Một hàm thuần túy không phụ thuộc vào trạng thái trước đó và luôn trả về cùng một kết quả cho cùng các tham số được truyền vào. Các chức năng thuần túy giúp tránh các vấn đề liên quan đến các đối tượng được chia sẻ, dữ liệu có thể thay đổi và các tác dụng phụ thường phổ biến trong môi trường đa luồng.

Lập trình phản ứng

Lập trình phản ứng đề cập đến lập trình theo hướng sự kiện trong đó các luồng dữ liệu đến theo kiểu không đồng bộ và được xử lý khi chúng đến.

Lập trình phản ứng chức năng

RxJava thực hiện cả hai khái niệm với nhau, trong đó dữ liệu của các luồng thay đổi theo thời gian và chức năng của người tiêu dùng phản ứng tương ứng.

Tuyên ngôn phản ứng

Tuyên ngôn phản ứng là một tài liệu trực tuyến nêu rõ tiêu chuẩn cao của hệ thống phần mềm ứng dụng. Theo tuyên ngôn, sau đây là các thuộc tính chính của một phần mềm phản ứng:

• Responsive - Nên luôn đáp ứng kịp thời.

• Message Driven - Nên sử dụng cách truyền thông điệp không đồng bộ giữa các thành phần để chúng duy trì mối ghép lỏng lẻo.

• Elastic - Luôn đáp ứng ngay cả khi chịu tải cao.

• Resilient - Luôn đáp ứng ngay cả khi bất kỳ (các) thành phần nào bị lỗi.

Các thành phần chính của RxJava

RxJava có hai thành phần chính: Observables và Observer.

• Observable - Nó đại diện cho một đối tượng tương tự như Stream có thể phát ra không hoặc nhiều dữ liệu, có thể gửi thông báo lỗi, có thể kiểm soát tốc độ của đối tượng này trong khi phát ra một tập dữ liệu, có thể gửi dữ liệu hữu hạn cũng như vô hạn.

• Observer- Nó đăng ký vào dữ liệu của Observable về trình tự và phản ứng trên mỗi mục của các có thể quan sát. Người quan sát được thông báo bất cứ khi nào Observable phát ra dữ liệu. Một Observer xử lý dữ liệu từng cái một.

Người quan sát không bao giờ được thông báo nếu các mục không có mặt hoặc một cuộc gọi lại không được trả lại cho một mục trước đó.

Thiết lập môi trường cục bộ

RxJava là một thư viện cho Java, vì vậy yêu cầu đầu tiên là phải cài đặt JDK trong máy của bạn.

Yêu cầu hệ thống

JDK	1,5 trở lên.
Ký ức	Không có yêu cầu tối thiểu.
Dung lượng đĩa	Không có yêu cầu tối thiểu.
Hệ điều hành	Không có yêu cầu tối thiểu.

Bước 1 - Xác minh cài đặt Java trong máy của bạn

Trước hết, mở bảng điều khiển và thực hiện một lệnh java dựa trên hệ điều hành bạn đang làm việc.

Hệ điều hành	Bài tập	Chỉ huy
các cửa sổ	Mở bảng điều khiển lệnh	c: \> java -version
Linux	Mở Command Terminal	$ java -version
Mac	Mở thiết bị đầu cuối	máy: <joseph $ java -version

Hãy xác minh kết quả đầu ra cho tất cả các hệ điều hành -

Hệ điều hành	Đầu ra
các cửa sổ	phiên bản java "1.8.0_101" Môi trường thời gian chạy Java (TM) SE (bản dựng 1.8.0_101)
Linux	phiên bản java "1.8.0_101" Môi trường thời gian chạy Java (TM) SE (bản dựng 1.8.0_101)
Mac	phiên bản java "1.8.0_101" Môi trường thời gian chạy Java (TM) SE (bản dựng 1.8.0_101)

Nếu bạn chưa cài đặt Java trên hệ thống của mình, hãy tải xuống Bộ phát triển phần mềm Java (SDK) từ liên kết sau https://www.oracle.com. Chúng tôi giả sử Java 1.8.0_101 là phiên bản đã cài đặt cho hướng dẫn này.

Bước 2 - Đặt môi trường JAVA

Đặt JAVA_HOMEbiến môi trường để trỏ đến vị trí thư mục cơ sở nơi Java được cài đặt trên máy của bạn. Ví dụ.

Hệ điều hành	Đầu ra
các cửa sổ	Đặt biến môi trường JAVA_HOME thành C: \ Program Files \ Java \ jdk1.8.0_101
Linux	xuất JAVA_HOME = / usr / local / java-current
Mac	export JAVA_HOME = / Library / Java / Home

Nối vị trí trình biên dịch Java vào Đường dẫn Hệ thống.

Hệ điều hành	Đầu ra
các cửa sổ	Nối chuỗi C:\Program Files\Java\jdk1.8.0_101\bin ở cuối biến hệ thống, Path.
Linux	xuất PATH = $ PATH: $ JAVA_HOME / bin /
Mac	không yêu cầu

Xác minh cài đặt Java bằng lệnh java -version Như đã giải thích ở trên.

Bước 3 - Tải xuống kho lưu trữ RxJava2

Tải xuống phiên bản mới nhất của tệp jar RxJava từ RxJava @ MVNRepository và Reactive Streams @ MVNRepository phụ thuộc của nó . Tại thời điểm viết hướng dẫn này, chúng tôi đã tải xuống rxjava-2.2.4.jar, react-Stream-1.0.2.jar và sao chép nó vào thư mục C: \> RxJava.

Hệ điều hành	Tên lưu trữ
các cửa sổ	rxjava-2.2.4.jar, react-stream-1.0.2.jar
Linux	rxjava-2.2.4.jar, react-stream-1.0.2.jar
Mac	rxjava-2.2.4.jar, react-stream-1.0.2.jar

Bước 4 - Đặt Môi trường RxJava

Đặt RX_JAVAbiến môi trường để trỏ đến vị trí thư mục cơ sở nơi lưu trữ jar RxJava trên máy của bạn. Giả sử chúng ta đã lưu trữ rxjava-2.2.4.jar và react-Stream-1.0.2.jar trong thư mục RxJava.

Sr.No	Hệ điều hành & Mô tả
1	Windows Đặt biến môi trường RX_JAVA thành C: \ RxJava
2	Linux xuất RX_JAVA = / usr / local / RxJava
3	Mac export RX_JAVA = / Library / RxJava

Bước 5 - Đặt biến CLASSPATH

Đặt CLASSPATH biến môi trường để trỏ đến vị trí jar RxJava.

Sr.No	Hệ điều hành & Mô tả
1	Windows Đặt biến môi trường CLASSPATH thành% CLASSPATH%;% RX_JAVA% \ rxjava-2.2.4.jar;% RX_JAVA% \ react-Stream-1.0.2.jar;.;
2	Linux export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: react-Stream-1.0.2.jar:.
3	Mac export CLASSPATH = $ CLASSPATH: $ RX_JAVA / rxjava-2.2.4.jar: react-Stream-1.0.2.jar:.

Bước 6 - Kiểm tra thiết lập RxJava

Tạo một lớp TestRx.java như hình dưới đây:

import io.reactivex.Flowable;
public class TestRx {
   public static void main(String[] args) {
      Flowable.just("Hello World!")
         .subscribe(System.out::println);
   }
}

Bước 7 - Xác minh kết quả

Biên dịch các lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac Tester.java

Xác minh kết quả đầu ra.

Hello World!

Observables đại diện cho các nguồn dữ liệu trong đó Observers (Subscribers)lắng nghe họ. Tóm lại, một Observable phát ra các mục và một Người đăng ký sau đó sẽ tiêu thụ các mục này.

Có thể quan sát được

• Observable cung cấp dữ liệu khi người đăng ký bắt đầu lắng nghe.

• Observable có thể phát ra bất kỳ số lượng mục nào.

• Có thể quan sát chỉ phát ra tín hiệu hoàn thành và không có mục nào.

• Observable có thể kết thúc thành công.

• Có thể quan sát không bao giờ kết thúc. ví dụ: một nút có thể được nhấp vào bất kỳ số lần nào.

• Có thể quan sát được lỗi bất kỳ lúc nào.

Người đăng kí

• Có thể quan sát được có nhiều người đăng ký.

• Khi một Observable phát ra một mục, mỗi phương thức onNext () của người đăng ký sẽ được gọi.

• Khi hoàn thành việc phát ra các mục có thể quan sát, mỗi phương thức onComplete () của người đăng ký sẽ được gọi.

• Nếu một Observable phát ra lỗi, mỗi phương thức onError () của người đăng ký sẽ được gọi.

Sau đây là các lớp cơ sở để tạo ra các có thể quan sát.

• Flowable- 0..N luồng, phát ra 0 hoặc n mục. Hỗ trợ dòng phản ứng và áp lực ngược.

• Observable - 0..N chảy, nhưng không có áp suất ngược.

• Single- 1 mục hoặc lỗi. Có thể được coi như một phiên bản phản ứng của cuộc gọi phương thức.

• Completable- Không có mục phát ra. Được sử dụng như một tín hiệu để hoàn thành hoặc lỗi. Có thể được coi như một phiên bản phản ứng của Runnable.

• MayBe- Không có mục nào hoặc phát ra 1 mục. Có thể được coi là phiên bản phản ứng của Tùy chọn.

Sau đây là các phương pháp thuận tiện để tạo các có thể quan sát trong lớp Quan sát được.

• just(T item) - Trả về một Observable báo hiệu mục đã cho (tham chiếu không đổi) và sau đó hoàn thành.

• fromIterable(Iterable source) - Chuyển đổi một chuỗi có thể lặp lại thành một nguồn quan sát có thể phát ra các mục trong chuỗi.

• fromArray(T... items) - Chuyển đổi một Mảng thành một nguồn quan sát có thể phát ra các mục trong Mảng.

• fromCallable(Callable supplier) - Trả về một Observable mà khi một người quan sát đăng ký vào nó, sẽ gọi một hàm mà bạn chỉ định và sau đó phát ra giá trị được trả về từ hàm đó.

• fromFuture(Future future) - Chuyển đổi Tương lai thành Nguồn quan sát.

• interval(long initialDelay, long period, TimeUnit unit) - Trả về một Observable phát ra 0L sau thời gian ban đầu và các số tăng dần sau mỗi khoảng thời gian sau đó.

Lớp Đơn đại diện cho phản hồi giá trị đơn. Có thể quan sát đơn lẻ chỉ có thể phát ra một giá trị thành công duy nhất hoặc một lỗi. Nó không phát ra sự kiện onComplete.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.Single<T> lớp học -

public abstract class Single<T>
   extends Object
      implements SingleSource<T>

Giao thức

Sau đây là giao thức tuần tự mà Single Observable hoạt động:

onSubscribe (onSuccess | onError)?

Ví dụ đơn

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {
      //Create the observable
      Single<String> testSingle = Single.just("Hello World");

      //Create an observer
      Disposable disposable = testSingle
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(
         new DisposableSingleObserver<String>() {

         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Hello World

Lớp MayBe đại diện cho phản hồi trì hoãn. MayBe có thể quan sát có thể phát ra một giá trị thành công duy nhất hoặc không có giá trị nào.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.Single<T> lớp học -

public abstract class Maybe<T>
   extends Object
      implements MaybeSource<T>

Giao thức

Sau đây là giao thức tuần tự mà MayBe Observable vận hành:

onSubscribe (onSuccess | onError | OnComplete)?

Ví dụ về MayBe

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Maybe;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableMaybeObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      //Create an observer
      Disposable disposable = Maybe.just("Hello World")
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(new DisposableMaybeObserver<String>() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }

         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Hello World

Lớp Hoàn thành đại diện cho phản hồi trì hoãn. Có thể quan sát được hoàn thành có thể cho biết hoàn thành thành công hoặc lỗi.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.Completable lớp học -

public abstract class Completable
extends Object
implements CompletableSource

Giao thức

Sau đây là giao thức tuần tự mà Completable Observable hoạt động:

onSubscribe (onError | onComplete)?

Ví dụ hoàn thành

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.concurrent.TimeUnit;

import io.reactivex.Completable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableCompletableObserver;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {

      //Create an observer
      Disposable disposable = Completable.complete()
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .subscribeWith(new DisposableCompletableObserver() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }
         @Override
         public void onStart() {
            System.out.println("Started!");
         }
         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 
      Thread.sleep(3000);
      //start observing
      disposable.dispose();
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Started!
Done!

Lớp CompositeDisposable đại diện cho một vùng chứa có thể chứa nhiều đồ dùng một lần và cung cấp độ phức tạp O (1) khi thêm và loại bỏ đồ dùng một lần.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.disposables.CompositeDisposable lớp học -

public final class CompositeDisposable
extends Object
implements Disposable, io.reactivex.internal.disposables.DisposableContainer

Ví dụ về CompositeDisposable

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Maybe;
import io.reactivex.Single;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableMaybeObserver;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {    
      CompositeDisposable compositeDisposable = new CompositeDisposable();

      //Create an Single observer 
      Disposable disposableSingle = Single.just("Hello World")
      .delay(2, TimeUnit.SECONDS, Schedulers.io())
      .subscribeWith(
      new DisposableSingleObserver<String>() {
         @Override
         public void onError(Throwable e) {
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }
      }); 

      //Create an observer
      Disposable disposableMayBe = Maybe.just("Hi")
      .delay(2, TimeUnit.SECONDS, Schedulers.io())
      .subscribeWith(new DisposableMaybeObserver<String>() {
         @Override
         public void onError(Throwable e) { 
            e.printStackTrace();
         }

         @Override
         public void onSuccess(String value) {
            System.out.println(value);
         }

         @Override
         public void onComplete() {
            System.out.println("Done!");
         }
      }); 

      Thread.sleep(3000);

      compositeDisposable.add(disposableSingle);
      compositeDisposable.add(disposableMayBe);

      //start observing
      compositeDisposable.dispose();
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Hello World
Hi

Sau đây là các toán tử được sử dụng để tạo một Observable.

Sr.No.	Nhà điều hành & Mô tả
1	Create Tạo một phương thức có thể quan sát từ đầu và cho phép phương thức quan sát viên gọi theo chương trình.
2	Defer Không tạo một Người quan sát cho đến khi một Người quan sát đăng ký. Tạo ra một quan sát mới cho mỗi người quan sát.
3	Empty/Never/Throw Tạo một Hành vi có thể quan sát được với hành vi hạn chế.
4	From Chuyển đổi đối tượng / cấu trúc dữ liệu thành có thể quan sát được.
5	Interval Tạo một số nguyên phát ra có thể quan sát theo trình tự với khoảng cách khoảng thời gian được chỉ định.
6	Just Chuyển đổi một đối tượng / cấu trúc dữ liệu thành một Có thể quan sát để phát ra cùng một loại hoặc cùng một loại đối tượng.
7	Range Tạo một số nguyên phát ra có thể quan sát theo trình tự của dải ô đã cho.
số 8	Repeat Tạo một số nguyên phát ra có thể quan sát được liên tiếp nhiều lần.
9	Start Tạo một Observable để phát ra giá trị trả về của một hàm.
10	Timer Tạo một mục có thể quan sát để phát ra một mục sau độ trễ nhất định.

Tạo ví dụ về toán tử

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using fromArray operator to create an Observable
public class ObservableTester  {
   public static void main(String[] args) { 
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .map(String::toUpperCase)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

ABCDEFG

Sau đây là các toán tử được sử dụng để biến đổi một mục được phát ra từ một Có thể quan sát.

Sr.No.	Nhà điều hành & Mô tả
1	Buffer Thu thập các mục từ Có thể quan sát thành các nhóm theo định kỳ và sau đó phát ra các nhóm chứ không phải các mục.
2	FlatMap Được sử dụng trong các đối tượng quan sát lồng nhau. Chuyển đổi các mục thành Có thể quan sát. Sau đó san phẳng các mục thành có thể quan sát duy nhất.
3	GroupBy Chia một Có thể quan sát thành tập hợp các Có thể quan sát được tổ chức theo khóa để tạo ra các nhóm mục khác nhau.
4	Map Áp dụng một chức năng cho mỗi mục được phát ra để biến đổi nó.
5	Scan Áp dụng một hàm cho từng mục được phát ra, tuần tự và sau đó phát ra giá trị kế tiếp.
6	Window Thu thập các mục từ Có thể quan sát vào cửa sổ Có thể quan sát theo định kỳ và sau đó phát ra các cửa sổ chứ không phải các mục.

Ví dụ về toán tử chuyển đổi

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using map operator to transform an Observable
public class ObservableTester  { 
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .map(String::toUpperCase)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

ABCDEFG

Sau đây là các toán tử được sử dụng để phát ra (các) mục có chọn lọc từ một Có thể quan sát.

Sr.No.	Nhà điều hành & Mô tả
1	Debounce Chỉ phát ra các mục khi hết thời gian chờ xảy ra mà không phát ra một mục khác.
2	Distinct Chỉ phát ra các mục duy nhất.
3	ElementAt chỉ phát ra mục ở chỉ số n do một Observable phát ra.
4	Filter Chỉ phát ra những mục vượt qua chức năng vị từ đã cho.
5	First Phát ra mục đầu tiên hoặc mục đầu tiên đã vượt qua tiêu chí đã cho.
6	IgnoreElements Không phát ra bất kỳ mục nào từ Có thể quan sát nhưng đánh dấu hoàn thành.
7	Last Phát ra phần tử cuối cùng từ Có thể quan sát.
số 8	Sample Phát ra mục gần đây nhất với khoảng thời gian nhất định.
9	Skip Bỏ qua n mục đầu tiên khỏi một Có thể quan sát.
10	SkipLast Bỏ qua n mục cuối cùng khỏi một Có thể quan sát.
11	Take lấy n mục đầu tiên từ một Observable.
12	TakeLast lấy n mục cuối cùng từ một Có thể quan sát.

Ví dụ về toán tử lọc

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using take operator to filter an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable
         .take(2)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

ab

Sau đây là các toán tử được sử dụng để tạo một Observable duy nhất từ ​​nhiều Observable.

Sr.No.	Nhà điều hành & Mô tả
1	And/Then/When Kết hợp các bộ mặt hàng bằng cách sử dụng các trung gian Mẫu và Kế hoạch.
2	CombineLatest Kết hợp mục mới nhất được phát ra bởi mỗi Observable thông qua một chức năng được chỉ định và phát ra mục kết quả.
3	Join Kết hợp các mục được phát ra bởi hai Có thể quan sát nếu được phát ra trong khung thời gian của mục phát ra Có thể quan sát thứ hai.
4	Merge Kết hợp các mục được phát ra từ các Quan sát.
5	StartWith Phát ra một chuỗi các mục cụ thể trước khi bắt đầu phát các mục từ nguồn có thể quan sát được
6	Switch Phát ra các mục gần đây nhất do Observables phát ra.
7	Zip Kết hợp các mục của Observables dựa trên chức năng và phát ra các mục kết quả.

Ví dụ về toán tử kết hợp

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using combineLatest operator to combine Observables
public class ObservableTester {
   public static void main(String[] args) {    
      Integer[] numbers = { 1, 2, 3, 4, 5, 6};
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable1 = Observable.fromArray(letters);
      Observable<Integer> observable2 = Observable.fromArray(numbers);
      Observable.combineLatest(observable1, observable2, (a,b) -> a + b)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

g1g2g3g4g5g6

Sau đây là các toán tử thường hữu ích với Observables.

Sr.No.	Nhà điều hành & Mô tả
1	Delay Đăng ký hành động để xử lý các sự kiện vòng đời có thể quan sát được.
2	Materialize/Dematerialize Đại diện cho mục được phát ra và thông báo được gửi.
3	ObserveOn Chỉ định bộ lập lịch được quan sát.
4	Serialize Buộc có thể quan sát để thực hiện các cuộc gọi tuần tự.
5	Subscribe Hoạt động dựa trên sự phát thải của các mục và thông báo như hoàn thành từ một thiết bị có thể quan sát được
6	SubscribeOn Chỉ định bộ lập lịch sẽ được sử dụng bởi một Observable khi nó được đăng ký.
7	TimeInterval Chuyển đổi một có thể quan sát để phát ra các chỉ báo về khoảng thời gian đã trôi qua giữa các lần phát xạ.
số 8	Timeout Đưa ra thông báo lỗi nếu xảy ra thời gian cụ thể mà không phát ra bất kỳ mục nào.
9	Timestamp Đính kèm dấu thời gian cho mỗi mục được phát ra.
9	Using Tạo tài nguyên dùng một lần hoặc cùng tuổi thọ với tài nguyên có thể quan sát được.

Ví dụ về toán tử tiện ích

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using subscribe operator to subscribe to an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable = Observable.fromArray(letters);
      observable.subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

abcdefg

Sau đây là các toán tử đánh giá một hoặc nhiều Observable hoặc các mục được phát ra.

Sr.No.	Nhà điều hành & Mô tả
1	All Đánh giá tất cả các mục được phát ra để đáp ứng các tiêu chí nhất định.
2	Amb Phát ra tất cả các mục từ Có thể quan sát đầu tiên chỉ được cung cấp nhiều Có thể quan sát.
3	Contains Kiểm tra xem một Observable có phát ra một mục cụ thể hay không.
4	DefaultIfEmpty Phát ra mục mặc định nếu Observable không phát ra bất kỳ thứ gì.
5	SequenceEqual Kiểm tra xem hai Bảng quan sát có phát ra cùng một chuỗi các mục không.
6	SkipUntil Loại bỏ các mục được phát ra bởi Có thể quan sát đầu tiên cho đến khi Có thể quan sát thứ hai tạo ra một mục.
7	SkipWhile Loại bỏ các mục do một Observable phát ra cho đến khi một điều kiện nhất định trở thành sai.
số 8	TakeUntil Loại bỏ các mục do một Observable phát ra sau khi một Observable thứ hai phát ra một mục hoặc kết thúc.
9	TakeWhile Loại bỏ các mục được phát ra bởi một Observable sau khi một điều kiện cụ thể trở thành sai.

Ví dụ về toán tử có điều kiện

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using defaultIfEmpty operator to operate on an Observable
public class ObservableTester  {
   public static void main(String[] args) {    
      final StringBuilder result = new StringBuilder();
      Observable.empty()
      .defaultIfEmpty("No Data")
      .subscribe(s -> result.append(s));
      System.out.println(result);
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result1 = new StringBuilder();
      Observable.fromArray(letters)
      .firstElement()
      .defaultIfEmpty("No data")   
      .subscribe(s -> result1.append(s));
      System.out.println(result1);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

No Data
a

Sau đây là các toán tử hoạt động trên toàn bộ các mục được phát ra bởi một Observable.

Sr.No.	Nhà điều hành & Mô tả
1	Average Đánh giá mức trung bình của tất cả các mục và đưa ra kết quả.
2	Concat Phát ra tất cả các mục từ nhiều Có thể quan sát được mà không cần xen kẽ.
3	Count Đếm tất cả các mục và đưa ra kết quả.
4	Max Đánh giá mục có giá trị tối đa của tất cả các mục và đưa ra kết quả.
5	Min Đánh giá mục có giá trị tối thiểu của tất cả các mục và đưa ra kết quả.
6	Reduce Áp dụng một chức năng trên mỗi mục và trả về kết quả.
7	Sum Đánh giá tổng của tất cả các mục và đưa ra kết quả.

Ví dụ về toán tử

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
//Using concat operator to operate on multiple Observables
public class ObservableTester  {
   public static void main(String[] args)  throws InterruptedException {    
      Integer[] numbers = { 1, 2, 3, 4, 5, 6};
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      Observable<String> observable1 = Observable.fromArray(letters);
      Observable<Integer> observable2 = Observable.fromArray(numbers);
      Observable.concat(observable1, observable2)
         .subscribe( letter -> result.append(letter));
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

abcdefg123456

Sau đây là các nhà khai thác có quyền kiểm soát chính xác hơn đối với việc đăng ký.

Sr.No.	Nhà điều hành & Mô tả
1	Connect Hướng dẫn một Observable có thể kết nối phát các mục cho người đăng ký của nó.
2	Publish Chuyển đổi một Observable thành Observable có thể kết nối.
3	RefCount Chuyển đổi một Observable có thể kết nối thành Observable thông thường.
4	Replay Đảm bảo mỗi người đăng ký có thể nhìn thấy cùng một chuỗi các mục đã phát, ngay cả sau khi Người quan sát đã bắt đầu phát các mục và người đăng ký đăng ký sau.

Ví dụ về toán tử có thể kết nối

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.observables.ConnectableObservable;
//Using connect operator on a ConnectableObservable
public class ObservableTester {
   public static void main(String[] args) {
      String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
      final StringBuilder result = new StringBuilder();
      ConnectableObservable<String> connectable = Observable.fromArray(letters).publish();      
      connectable.subscribe(letter -> result.append(letter));
      System.out.println(result.length());
      connectable.connect();
      System.out.println(result.length());
      System.out.println(result);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

0
7
abcdefg

Theo Reactive, một Chủ thể có thể hoạt động như cả Người quan sát và Người quan sát.

Chủ thể là một loại cầu nối hoặc proxy có sẵn trong một số triển khai của ReactiveX, hoạt động như một người quan sát và một người có thể quan sát được. Bởi vì nó là một người quan sát, nó có thể đăng ký một hoặc nhiều Quan sát và vì nó là một Người quan sát, nó có thể đi qua các mục mà nó quan sát bằng cách lắp lại chúng và nó cũng có thể phát ra các mục mới.

Có bốn loại Chủ thể -

Sr.No.	Miêu tả môn học
1	Publish Subject Chỉ phát ra những mục được phát sau thời gian đăng ký.
2	Replay Subject Phát ra tất cả các mục được phát ra bởi nguồn Observable bất kể nó đã đăng ký Observable.
3	Behavior Subject Sau khi đăng ký, phát mục gần đây nhất, sau đó tiếp tục phát mục do nguồn có thể quan sát được.
4	Async Subject Phát ra mục cuối cùng được phát ra bởi nguồn có thể quan sát được sau khi hoàn thành phát.

PublishSubject phát các mục cho Người quan sát hiện đã đăng ký và các sự kiện đầu cuối cho Người quan sát hiện tại hoặc muộn.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.subjects.PublishSubject<T> lớp học -

public final class PublishSubject<T>
extends Subject<T>

Ví dụ về PublishSubject

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.PublishSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      PublishSubject<String> subject = PublishSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be abcd 
      System.out.println(result1);
      //Output will be d only
      //as subscribed after c item emitted.
      System.out.println(result2);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

abcd
d

BehaviorSubject phát ra mục gần đây nhất mà nó đã quan sát và sau đó tất cả các mục được quan sát tiếp theo cho mỗi Người quan sát đã đăng ký.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.subjects.BehaviorSubject<T> lớp học -

public final class BehaviorSubject<T>
extends Subject<T>

Ví dụ về BehaviorSubject

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.BehaviorSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         
      BehaviorSubject<String> subject =  BehaviorSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();
      //Output will be abcd
      System.out.println(result1);
      //Output will be cd being BehaviorSubject 
      //(c is last item emitted before subscribe)
      System.out.println(result2);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

abcd
cd

ReplaySubject phát lại các sự kiện / mục cho Người quan sát hiện tại và cuối cùng.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.subjects.ReplaySubject<T> lớp học -

public final class ReplaySubject<T>
extends Subject<T>

Ví dụ về ReplaySubject

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects.ReplaySubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      ReplaySubject<String> subject = ReplaySubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be abcd
      System.out.println(result1);
      //Output will be abcd being ReplaySubject
      //as ReplaySubject emits all the items
      System.out.println(result2);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

abcd
abcd

AsyncSubject phát ra giá trị cuối cùng duy nhất theo sau là sự kiện hoàn thành hoặc lỗi nhận được cho Người quan sát.

Khai báo lớp học

Sau đây là khai báo cho io.reactivex.subjects.AsyncSubject<T> lớp học -

public final class  AsyncSubject<T>
extends Subject<T>

Ví dụ về AsyncSubject

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.subjects. AsyncSubject;
public class ObservableTester  {
   public static void main(String[] args) {   
      final StringBuilder result1 = new StringBuilder();
      final StringBuilder result2 = new StringBuilder();         

      AsyncSubject<String> subject =  AsyncSubject.create(); 
      subject.subscribe(value -> result1.append(value) ); 
      subject.onNext("a"); 
      subject.onNext("b"); 
      subject.onNext("c"); 
      subject.subscribe(value -> result2.append(value)); 
      subject.onNext("d"); 
      subject.onComplete();

      //Output will be d being the last item emitted
      System.out.println(result1);
      //Output will be d being the last item emitted     
      System.out.println(result2);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

d
d

Bộ lập lịch được sử dụng trong môi trường đa luồng để làm việc với các toán tử có thể quan sát.

Theo Reactive, Scheduler được sử dụng để lập lịch trình chuỗi các nhà khai thác sẽ áp dụng như thế nào cho các luồng khác nhau.

Theo mặc định, một Observable và chuỗi toán tử mà bạn áp dụng cho nó sẽ thực hiện công việc của nó và sẽ thông báo cho những người quan sát của nó, trên cùng một chuỗi mà phương thức Subscribe của nó được gọi. Toán tử SubscribeOn thay đổi hành vi này bằng cách chỉ định một Bộ lập lịch khác mà Observable sẽ hoạt động. Toán tử ObserOn chỉ định một Bộ lập lịch khác mà Người quan sát sẽ sử dụng để gửi thông báo cho người quan sát của nó.

Có các loại Bộ lập lịch sau đây có sẵn trong RxJava:

Sr.No.	Trình lập lịch & Mô tả
1	Schedulers.computation() Tạo và trả về một Bộ lập lịch dành cho công việc tính toán. Số luồng được lên lịch phụ thuộc vào các CPU có trong hệ thống. Một luồng được phép cho mỗi CPU. Tốt nhất cho các vòng lặp sự kiện hoặc hoạt động gọi lại.
2	Schedulers.io() Tạo và trả về một Trình lập lịch dành cho công việc có IO. Nhóm chủ đề có thể mở rộng khi cần thiết.
3	Schedulers.newThread() Tạo và trả về một Bộ lập lịch trình tạo một Chuỗi mới cho mỗi đơn vị công việc.
4	Schedulers.trampoline() Tạo và trả về một Bộ lập lịch hàng đợi hoạt động trên luồng hiện tại sẽ được thực thi sau khi công việc hiện tại hoàn thành.
4	Schedulers.from(java.util.concurrent.Executor executor) Chuyển đổi Người thực thi thành một phiên bản Người lập lịch mới.

Phương thức Schedulers.trampoline () tạo và trả về một Scheduler xếp hàng đợi hoạt động trên luồng hiện tại sẽ được thực thi sau khi công việc hiện tại hoàn thành.

Ví dụ về Schedulers.trampoline ()

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.trampoline()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Processing Thread main
Receiver Thread main, Item length 1
Processing Thread main
Receiver Thread main, Item length 2
Processing Thread main
Receiver Thread main, Item length 3

Phương thức Schedulers.newThread () tạo và trả về một Scheduler để tạo một Thread mới cho mỗi đơn vị công việc.

Ví dụ về Schedulers.newThread ()

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.newThread()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Processing Thread RxNewThreadScheduler-1
Receiver Thread RxNewThreadScheduler-1, Item length 1
Processing Thread RxNewThreadScheduler-2
Receiver Thread RxNewThreadScheduler-2, Item length 2
Processing Thread RxNewThreadScheduler-3
Receiver Thread RxNewThreadScheduler-3, Item length 3

Phương thức Schedulers.computation () tạo và trả về một Scheduler dành cho công việc tính toán. Số luồng được lên lịch phụ thuộc vào các CPU có trong hệ thống. Một luồng được phép cho mỗi CPU. Tốt nhất cho các vòng lặp sự kiện hoặc hoạt động gọi lại.

Ví dụ về Schedulers.computation ()

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.computation()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Processing Thread RxComputationThreadPool-1
Receiver Thread RxComputationThreadPool-1, Item length 1
Processing Thread RxComputationThreadPool-2
Receiver Thread RxComputationThreadPool-2, Item length 2
Processing Thread RxComputationThreadPool-3
Receiver Thread RxComputationThreadPool-3, Item length 3

Phương thức Schedulers.io () tạo và trả về một Scheduler dành cho công việc ràng buộc IO. Nhóm chủ đề có thể mở rộng khi cần thiết. Tốt nhất cho I / O hoạt động chuyên sâu.

Ví dụ về Schedulers.io ()

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.Random;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.io()))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 1
Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 2
Processing Thread RxCachedThreadScheduler-1
Receiver Thread RxCachedThreadScheduler-1, Item length 3

Phương thức Schedulers.from (Executor) chuyển đổi một Executor thành một thể hiện Scheduler mới.

Ví dụ về Schedulers.from (Executor)

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import java.util.Random;
import java.util.concurrent.Executors;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class ObservableTester  {
   public static void main(String[] args) throws InterruptedException {
      Observable.just("A", "AB", "ABC")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.from(Executors.newFixedThreadPool(3))))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Processing Thread pool-1-thread-1
Processing Thread pool-3-thread-1
Receiver Thread pool-1-thread-1, Item length 1
Processing Thread pool-4-thread-1
Receiver Thread pool-4-thread-1, Item length 3
Receiver Thread pool-3-thread-1, Item length 2

Toán tử đệm cho phép tập hợp các mục được phát ra bởi một Observable thành một danh sách hoặc các gói và phát ra các gói đó thay vì các mục. Trong ví dụ dưới đây, chúng tôi đã tạo một Observable để phát ra 9 mục và sử dụng bộ đệm, 3 mục sẽ được phát ra cùng nhau.

Ví dụ về đệm

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      Observable<Integer> observable = Observable.just(1, 2, 3, 4,
         5, 6, 7, 8, 9);

      observable.subscribeOn(Schedulers.io())
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .buffer(3)
         .subscribe(new Observer<List<Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
               System.out.println("Subscribed");
            }
            @Override
            public void onNext(List<Integer> integers) {
               System.out.println("onNext: ");
               for (Integer value : integers) {
                  System.out.println(value);
               }
            }
            @Override
            public void onError(Throwable e) {
               System.out.println("Error");
            }

            @Override
            public void onComplete() {
               System.out.println("Done! ");
            }
         });
      Thread.sleep(3000);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Subscribed
onNext: 
1
2
3
onNext: 
4
5
6
onNext: 
7
8
9
Done!

Toán tử Windowing hoạt động tương tự như toán tử bộ đệm nhưng nó cho phép tập hợp các mục do một Observable phát ra vào một Observable khác thay vì tập hợp và phát ra Observable đó thay vì tập hợp. Trong ví dụ dưới đây, chúng tôi đã tạo một Observable để phát ra 9 mục và sử dụng toán tử cửa sổ, 3 Observable sẽ được phát ra cùng nhau.

Ví dụ về gió

Tạo chương trình Java sau đây bằng cách sử dụng bất kỳ trình soạn thảo nào bạn chọn trong C: \> RxJava.

ObservableTester.java

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class ObservableTester {
   public static void main(String[] args) throws InterruptedException {
      Observable<Integer> observable = Observable.just(1, 2, 3, 4,
         5, 6, 7, 8, 9);

      observable.subscribeOn(Schedulers.io())
         .delay(2, TimeUnit.SECONDS, Schedulers.io())
         .window(3)
         .subscribe(new Observer<Observable<Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
               System.out.println("Subscribed");
            }
            @Override
            public void onNext(Observable<Integer> integers) {
               System.out.println("onNext: ");
               integers.subscribe(value -> System.out.println(value));
            }
            @Override
            public void onError(Throwable e) {
               System.out.println("Error");
            }

            @Override
            public void onComplete() {
               System.out.println("Done! ");
            }
         });
      Thread.sleep(3000);
   }
}

Xác minh kết quả

Biên dịch lớp bằng cách sử dụng javac trình biên dịch như sau -

C:\RxJava>javac ObservableTester.java

Bây giờ hãy chạy ObservableTester như sau:

C:\RxJava>java ObservableTester

Nó sẽ tạo ra kết quả sau:

Subscribed
onNext: 
1
2
3
onNext: 
4
5
6
onNext: 
7
8
9
Done!