2. 程式人生 > >RxJava 併發之執行緒排程

RxJava 併發之執行緒排程

阿新 • • 發佈：2019-01-10

由於 Rx 目標是用在非同步系統上並且 Rx 支援多執行緒處理，所以很多 Rx 開發者認為預設情況下 Rx 就是多執行緒的。 其實實際情況不是這樣的，Rx 預設是單執行緒的。

除非你明確的指定執行緒，否則所有 onNext/onError/onCompleted 以及各個操作函式的呼叫都是在同一個執行緒中完成的。例如下面的示例：

final BehaviorSubject<Integer> subject = BehaviorSubject.create();
subject.subscribe(i -> {
    System.out.println("Received "
 
 + i + " on " + Thread.currentThread().getId());
});

int[] i = {1}; // naughty side-effects for examples only ;)
Runnable r = () -> {
    synchronized(i) {
        System.out.println("onNext(" + i[0] + ") on " + Thread.currentThread().getId());
        subject.onNext(i[0]++);
    }
};

r.run(); // Execute on main thread
 

new Thread(r).start();
new Thread(r).start();

結果：

onNext(1) on 1
Received 1 on 1
onNext(2) on 11
Received 2 on 11
onNext(3) on 12
Received 3 on 12

上面在三個執行緒中分別呼叫 subject 的onNext 函式。和 Runnable 中的執行緒是同一個執行緒。不管用多少個操作函式串聯呼叫，結果都是同一個執行緒。

subscribeOn 和 observeOn

subscribeOn 和 observeOn 分別用來控制 subscription 的呼叫執行緒和 接受事件通知（Observer 的 onNext/onError/onCompleted 函式）的執行緒。

public final Observable<T> observeOn(Scheduler scheduler)
public final Observable<T> subscribeOn(Scheduler scheduler)

在Rx 中你並不直接和 執行緒 打交道，而是通過 Scheduler 來處理多執行緒。

subscribeOn

subscribeOn 用來指定 Observable.create 中的程式碼在那個 Scheduler 中執行。即使你沒有呼叫 create 函式，但是內部也有一個 create 實現。例如：

System.out.println("Main: " + Thread.currentThread().getId());

Observable.create(o -> {
        System.out.println("Created on " + Thread.currentThread().getId());
        o.onNext(1);
        o.onNext(2);
        o.onCompleted();
    })
    //.subscribeOn(Schedulers.newThread())
    .subscribe(i -> {
        System.out.println("Received " + i + " on " + Thread.currentThread().getId());
    });

System.out.println("Finished main: " + Thread.currentThread().getId());

結果：

Main: 1
Created on 1
Received 1 on 1
Received 2 on 1
Finished main: 1

可以看到上面的程式碼是在同一個執行緒中執行，並且是按循序執行的。subscribe 執行完後（包括create 函式裡面的 Lambda 表示式的程式碼）才繼續執行後面的程式碼。

如果你把上面的註釋掉的程式碼 .subscribeOn(Schedulers.newThread()) 啟用，這結果是這樣的：

Main: 1
Finished main: 1
Created on 11
Received 1 on 11
Received 2 on 11

這樣 create 裡面的 Lambda 表示式程式碼將會在 Schedulers.newThread() 返回的執行緒中執行。subscribe 不再是阻塞的了。後面的程式碼可以立即執行，而不用等待 subscribe 返回。

有些 Observable 內部會使用它們自己建立的執行緒。例如 Observable.interval 就是非同步的。這種情況下，無需指定新的執行緒。

System.out.println("Main: " + Thread.currentThread().getId());

Observable.interval(100, TimeUnit.MILLISECONDS)
    .subscribe(i -> {
        System.out.println("Received " + i + " on " + Thread.currentThread().getId());
    });

System.out.println("Finished main: " + Thread.currentThread().getId());

結果：

Main: 1
Finished main: 1
Received 0 on 11
Received 1 on 11
Received 2 on 11

observeOn

observeOn 控制資料流的另外一端。你的 observer 如何收到事件。也就是在那個執行緒中回撥 observer 的 onNext/onError/onCompleted 函式。

Observable.create(o -> {
        System.out.println("Created on " + Thread.currentThread().getId());
        o.onNext(1);
        o.onNext(2);
        o.onCompleted();
    })
    .observeOn(Schedulers.newThread())
    .subscribe(i ->
        System.out.println("Received " + i + " on " + Thread.currentThread().getId()));

結果：

Created on 1
Received 1 on 13
Received 2 on 13

observeOn 隻影響呼叫該函式以後的操作函式。你可以認為 observeOn 只是攔截了資料流並且對後續的操作有作用。例如：

Observable.create(o -> {
        System.out.println("Created on " + Thread.currentThread().getId());
        o.onNext(1);
        o.onNext(2);
        o.onCompleted();
    })
    .doOnNext(i -> 
        System.out.println("Before " + i + " on " + Thread.currentThread().getId()))
    .observeOn(Schedulers.newThread())
    .doOnNext(i -> 
        System.out.println("After " + i + " on " + Thread.currentThread().getId()))
    .subscribe();

結果：

Created on 1
Before 1 on 1
Before 2 on 1
After 1 on 13
After 2 on 13

可以看到在遇到 observeOn 之前，所有的操作發生在一個執行緒，之後在另外一個執行緒。這樣可以在 Rx 資料流中不同地方設定不同的執行緒。

如果你知道資料流處理在那些情況需要很長時間，則可以通過這個操作來避免阻塞生產者執行緒。 比如在 Android 開發過程中的 UI 執行緒，如果在該執行緒中讀取檔案，可能會導致 UI 卡死（ANR）無響應，通過該函式可以指定讀取檔案在另外一個執行緒中執行。

unsubscribeOn

有些 Observable 會依賴一些資源，當該 Observable 完成後釋放這些資源。如果釋放資源比較耗時的話，可以通過 unsubscribeOn 來指定 釋放資原始碼執行的執行緒。

Observable<Object> source = Observable.using(
    () -> {
        System.out.println("Subscribed on " + Thread.currentThread().getId());
        return Arrays.asList(1,2);
    },
    (ints) -> {
        System.out.println("Producing on " + Thread.currentThread().getId());
        return Observable.from(ints);
    },
    (ints) -> {
        System.out.println("Unubscribed on " + Thread.currentThread().getId());
    }
);

source
    .unsubscribeOn(Schedulers.newThread())
    .subscribe(System.out::println);

結果：

Subscribed on 1
Producing on 1
1
2
Unubscribed on 11

Schedulers

observeOn 和 subscribeOn 的引數為一個 Scheduler 物件。Scheduler 是用來協調任務執行的。 RxJava 包含了一些常用的 Scheduler，你也可以自定義 Scheduler。 通過呼叫 Schedulers 的工廠函式來獲取標準的預定義的 Scheduler。

RxJava 內建的 Scheduler 有：

  • immediate 同步執行
  • trampoline 把任務放到當前執行緒的佇列中，等當前任務執行完了，再繼續執行佇列中的任務
  • newThread 對於每個任務建立一個新的執行緒去執行
  • computation 計算執行緒，用於需要大量 CPU 計算的任務
  • io 用於執行 io 操作的任務
  • test 用於測試和除錯

當前 computation 和 io 的實現是類似的，他們兩個主要用來確保呼叫的場景，相當於文件說明，來表明你當前的任務是何種型別的。

大部分的 Rx 操作函式內部都使用了schedulers 。並且大部分的 Observable 操作函式也都有一個使用 Scheduler 引數的過載函式。通過過載函式可以指定該操作函式執行的執行緒。

scheduler 的高階特性

Rx scheduler 的使用場景並沒有限定在 Rx 中，也可以在普通 Java 程式碼中使用。

執行一個任務

Scheduler 有個 createWorker 函式，用來建立一個可以執行的任務（Scheduler.Worker）。然後可以排程該任務：

Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(
    () -> System.out.println("Action"));

上面的任務被分配到其指定的執行緒中了。

還可以重複執行任務，或者只執行一次，也可以推遲任務執行：

Subscription schedule(
    Action0 action,
    long delayTime,
    java.util.concurrent.TimeUnit unit)
Subscription schedulePeriodically(
    Action0 action,
    long initialDelay,
    long period,
    java.util.concurrent.TimeUnit unit)
Scheduler scheduler = Schedulers.newThread();
long start = System.currentTimeMillis();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(
    () -> System.out.println(System.currentTimeMillis()-start),
    5, TimeUnit.SECONDS);
worker.schedule(
    () -> System.out.println(System.currentTimeMillis()-start),
    5, TimeUnit.SECONDS);

結果：

5033
5035

上面示例中可以看到，推遲執行是從排程開始的時候計算時間的。

取消任務

Scheduler.Worker 繼承至 Subscription。呼叫 unsubscribe 函式可以取消佇列中的任務：

Scheduler scheduler = Schedulers.newThread();
long start = System.currentTimeMillis();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(
    () -> {
        System.out.println(System.currentTimeMillis()-start);
        worker.unsubscribe();
    },
    5, TimeUnit.SECONDS);
worker.schedule(
    () -> System.out.println(System.currentTimeMillis()-start),
    5, TimeUnit.SECONDS);

結果：

5032

第一個任務中呼叫了 unsubscribe，這樣第二個任務被取消了。下面一個示例演示任務沒有執行完，被取消的情況，會丟擲一個 InterruptedException 異常：

Scheduler scheduler = Schedulers.newThread();
long start = System.currentTimeMillis();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
    try {
        Thread.sleep(2000);
        System.out.println("Action completed");
    } catch (InterruptedException e) {
        System.out.println("Action interrupted");
    }
});
Thread.sleep(500);
worker.unsubscribe();

結果：

Action interrupted

schedule 返回的是一個 Subscription 物件，可以在該物件上呼叫取消操作，這樣可以只取消這一個任務，而不是取消所有任務。

RxJava 中現有的 scheduler

ImmediateScheduler

ImmediateScheduler 並沒有做任何執行緒排程。只是同步的執行任務。巢狀呼叫會導致任務被遞迴執行：

Scheduler scheduler = Schedulers.immediate();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
    System.out.println("Start");
    worker.schedule(() -> System.out.println("Inner"));
    System.out.println("End");
});

結果：

Start
Inner
End

TrampolineScheduler

TrampolineScheduler 也是同步執行，但是不巢狀任務。而是把後來的任務新增到任務佇列中，等前面的任務執行完了 再執行後面的。

Scheduler scheduler = Schedulers.trampoline();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
    System.out.println("Start");
    worker.schedule(() -> System.out.println("Inner"));
    System.out.println("End");
});

結果：

Start
End
Inner

TrampolineScheduler 把任務安排到第一次執行任務的那個執行緒中執行。這樣，第一次呼叫 schedule 的操作是阻塞的，直到佇列執行完。後續的任務，會在這個執行緒中一個一個的執行，並且後續的呼叫不會阻塞。

NewThreadScheduler

NewThreadScheduler 給每個任務建立一個新的執行緒。
定義一個列印執行緒資訊的輔助函式：

public static void printThread(String message) {
    System.out.println(message + " on " + Thread.currentThread().getId());
}

示例：

printThread("Main");
Scheduler scheduler = Schedulers.newThread();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
    printThread("Start");
    worker.schedule(() -> printThread("Inner"));
    printThread("End");
});
Thread.sleep(500);
worker.schedule(() -> printThread("Again"));

結果：

Main on 1
Start on 11
End on 11
Inner on 11
Again on 11

相關推薦

RxJava 併發之執行緒排程

由於 Rx 目標是用在非同步系統上並且 Rx 支援多執行緒處理，所以很多 Rx 開發者認為預設情況下 Rx 就是多執行緒的。 其實實際情況不是這樣的，Rx 預設是單執行緒的。 除非你明確的指定執行緒，否則所有 onNext/onError/onComplete

Java多執行緒-併發之執行緒池

執行緒池有了解嗎？ 答： java.util.concurrent.ThreadPoolExecutor 類就是一個執行緒池。客戶端呼叫ThreadPoolExecutor.submit(Runnable task) 提交任務，執行緒池內部維護的工作者執行緒的數量就是該執行緒池的執行

Java多執行緒-併發之執行緒和程序的區別

執行緒和程序的區別 答： 程序是一個“執行中的程式”，是系統進行資源分配和排程的一個獨立單位 執行緒是程序的一個實體，一個程序中擁有多個執行緒，執行緒之間共享地址空間和其他資源（所以通訊和同步等操作執行緒比程序更加容易） 執行緒上下文的切換比程序上下文切換要快

Java多執行緒之執行緒排程（二）

（一）執行緒優先順序 執行緒優先順序用1~10表示，10表示優先順序最高，預設值是5.每個優先順序對應一個Thread類的公用靜態常量。如 public static final int MIN_PRIORITY = 1; public static final int NO

java併發之----執行緒的建立方法

一、執行緒的建立 （1）繼承Thread類 需要實現run方法，通過呼叫start方法啟動執行緒 public class MyThread extends Thread{ @Override public void run(){//實現run方法 //以下根據自己的需要

Java多執行緒之執行緒排程詳解

排程的概念 給定時間結點，時間間隔，任務次數，然後自動執行任務 應用場景舉例 1.日誌清理：每隔三個月，清理公司日誌檔案 2.日誌備份：每個一週，備份公司檔案 3.工資結算：每個月29號，考勤彙報，業務結算，計算工資 排程的實現方式：

python 併發之執行緒

一.什麼是執行緒 #指的是一條流水線的工作過程，關鍵的一句話：一個程序內最少自帶一個執行緒，其實程序根本不能執行，程序不是執行單位，是資源的單位，分配資源的單位 #執行緒才是執行單位 #程序：做手機螢幕的工作過程,剛才講的 #我們的py檔案在執行的時候，如果你站在資源單位的角度來看，我

RT-Thread核心之執行緒排程(六)

5.執行緒切換的本質？ 到現在我們知道，每個執行緒的執行需要一定的“物質”基礎。首先，需要獲得CPU的使用權，這就包括CPU內部各暫存器的使用，然後有自己獨立的棧空間，這部分的空間每個執行緒應該各自獨

Java併發之執行緒封閉

讀者們好！ 在這篇部落格中，我們將探討執行緒封閉是什麼意思，以及我們如何實現它。 所以，讓我們直接開始吧。 1. 執行緒封閉 大多數的併發問題僅發生在我們想要線上程之間共享可變變數或可變狀態時。如果在多個執行緒之間操作共享變數，則所有執行緒都將能夠讀取和修改變數的值，從而出現意外或不正確的結果。一種簡單的

解密TaurusDB儲存端高併發之執行緒池

摘要：為了能加快相關任務的高效執行，TaurusDB採用多執行緒技術處理的方式，增加處理器單元的吞吐能力，從而提高儲存端的執行效率。 1. TaurusDB背景 隨著雲端計算進入2.0時代，資料急劇膨脹，這對實現資料庫的高可靠、高效能、高吞吐的目標產生了巨大的挑戰。如圖1 所示，TaurusDB是華為自研的最

IIS 之 連線數、併發連線數、最大併發工作執行緒數、佇列長度、最大工作程序數

轉載： 參考：https://blog.csdn.net/enweitech/article/details/79815137 https://blog.csdn.net/enweitech/article/details/79815137 https://blog.csdn.net/b

Python併發程式設計之執行緒池/程序池

Python併發程式設計之執行緒池/程序池 2017/01/18 · 基礎知識 · 2 評論 · 併發, 執行緒池, 程序池 原文出處： ZiWenXie    引言 Pyt

Java併發程式設計之執行緒生命週期、守護執行緒、優先順序和join、sleep、yield

Java併發程式設計中，其中一個難點是對執行緒生命週期的理解，和多種執行緒控制方法、執行緒溝通方法的靈活運用。這些方法和概念之間彼此聯絡緊密，共同構成了Java併發程式設計基石之一。 Java執行緒的生命週期 Java執行緒類定義了New、Runnable、Running Man、Blocked和Dead

Java併發程式設計之執行緒安全、執行緒通訊

Java多執行緒開發中最重要的一點就是執行緒安全的實現了。所謂Java執行緒安全，可以簡單理解為當多個執行緒訪問同一個共享資源時產生的資料不一致問題。為此，Java提供了一系列方法來解決執行緒安全問題。 synchronized synchronized用於同步多執行緒對共享資源的訪問，在實現中分為同步程

java多執行緒系列翻譯之java併發/多執行緒教程

原文地址：http://tutorials.jenkov.com/java-concurrency/index.html 以前計算機都是單核，同時只能執行一個程式。之後出現了多重任務處理，這意味著計算機同時可以處理多個程式（又名任務或流程）。但這不是真正的“同時執行”，只是單個CPU被多個程式共

Java併發程式設計(5)-Executor執行緒排程框架解讀

文章目錄 一、Executor執行緒排程框架 1.1、什麼是執行緒排程框架 1.2、Executors 1.3、Executor 1.4、ExecutorService 1

Java併發程式設計之執行緒池(三)

一.介紹 Java通過Executors提供四種執行緒池，分別為： (1)newCachedThreadPool:建立一個可快取執行緒池，如果執行緒池長度超過處理需要，可靈活回收空閒執行緒，若無可回收，則新建執行緒。 (2)newFixedThreadPool: 建立一個定長執行緒池，可控制

java併發程式設計之執行緒的基本概念

本文為學習筆記。源自學習微信公眾號“我們都是小青蛙”。 本篇文章將記錄如何使用java中的執行緒。 main執行緒 main方法是程式入口，我們對已經編譯好的class檔案呼叫java命令時就可以執行一個java程式。這個過程中，其實系統自動為我們建立了一個程序

《java併發程式設計實戰》之 執行緒安全性

1.執行緒安全性 當多個執行緒訪問某個類時，不管執行時環境採用何種排程方式或者這些執行緒將如何交替執行，並且在主調程式碼中不需要任何額外的同步或協同，這個類都能表現出正確的行為，那麼這個類就是執行緒安全的。 無狀態物件一定是執行緒安全的，何為無狀態，就是類中不包含任何域，也不包含各種其

Java學習筆記之--------執行緒任務排程

Timer定時類：一種工具，執行緒用其安排以後在後臺執行緒中執行的任務。可安排任務執行一次，或者定期重複執行。 schedule(TimerTask task，Date time)：安排在指定的時間執行指定的任務。 schedule(TimerTask task，Date firstTime，long pe