RxJava3.0 入門教程

一、說明

rxjava 是基於事件的異步編程。在寫代碼之前，我們首先要了解幾個概念。  （如果有什么錯誤之處，還請指正）

• Observable(被觀察者，可觀察對象，就是要進行什么操作，相當於生產者)
• bscriber 負責處理事件，他是事件的消費者
• Operator 是對 Observable 發出的事件進行修改和變換
• subscribeOn(): 指定 subscribe() 所發生的線程
• observeOn(): 指定 Subscriber 所運行在的線程。或者叫做事件消費的線程

RxJava 3具有幾個基礎類

• io.reactivex.rxjava3.core.Flowable：0..N流量，支持反應流和背壓
• io.reactivex.rxjava3.core.Observable：0..N流動，無背壓，
• io.reactivex.rxjava3.core.Single：正好是1個項目的流或一個錯誤，
• io.reactivex.rxjava3.core.Completable：沒有項目但只有完成或錯誤信號的流程，
• io.reactivex.rxjava3.core.Maybe：沒有項目，恰好一項或錯誤的流程。

 

二、代碼示例

1. 一個簡單代碼示例

public class Main {

    public static void main(String[] args) {

        Flowable.just("hello world ").subscribe(System.out::println);
        System.out.println("結束");
        
    }
}

// 結果：
hello world 
結束

2.  結束是打印在后面

public static void main(String[] args) {
        Flowable.range(0,5).map(x->x * x).subscribe(x->{
            Thread.sleep(1000);
            System.out.println(x);
        });
        System.out.println("結束");
    }

3.   可設置生產者和消費者使用的線程模型 。

public static void main(String[] args) throws Exception{
        Flowable<String> source = Flowable.fromCallable(() -> {
            Thread.sleep(1000); //  imitate expensive computation
            return "Done";
        });

        Flowable<String> runBackground = source.subscribeOn(Schedulers.io());
        Flowable<String> showForeground = runBackground.observeOn(Schedulers.single());
        showForeground.subscribe(System.out::println, Throwable::printStackTrace);
        System.out.println("------");
        Thread.sleep(2000);
        source.unsubscribeOn(Schedulers.io());//事件發送完畢后，及時取消發送
        System.out.println("結束");
    }

4.   發現異步效果      是並行執行的結果

Flowable.range(1, 10)
                .observeOn(Schedulers.single())
                .map(v -> v * v)
                .subscribe(System.out::println);
        System.out.println("結束");

5. 無異步效果

Flowable.range(1, 10)
                .observeOn(Schedulers.single())
                .map(v -> v * v)
                .blockingSubscribe(System.out::println);
        System.out.println("結束");