2. 程式人生 > >4. Other Observable sources

4. Other Observable sources

阿新 • • 發佈：2018-12-21

Observable.range()

可以使用Observable.range()發射一定範圍內的整數。從start值開始傳送發射，每次加1,直到一定的count。這些數通過onNext()事件傳遞，跟著一個onComplete()事件：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
        Observable.range(1,10)
            .subscribe(s -> System.out.println
 
("RECEIVED: " + s));
    }
}

可以使用Observable.rangeLong()發射比較大的數。

Observable.interval()

Observable.interval()生成一個基於時間的Observable。每經過一個時間間隔它將發射一個連續的long emission（從0開始）。 下面的程式碼，每秒鐘發射一個數：

import io.reactivex.Observable;
import java.util.concurrent.TimeUnit;

public class Launcher {
    public static void
 
 main(String[]args) {
        Observable.interval(1, TimeUnit.SECONDS)
            .subscribe(s -> System.out.println(s + " Mississippi"));
        sleep(5000);
    }
    
    public static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException
 
 e) {
            e.printStackTrace();
        }
    }
}

Observable.interval()在一個定時器上執行，所以，需要一個單獨的執行緒做排程。上面的程式，main()方法生成一個Observable，但是不等它完成。它在一個新的執行緒上發射。要讓main()不退出，我們使用sleep()方法讓程式存活了5秒鐘。在程式退出之前，我們的Observable有5秒鐘時間可以發射。當你增加一個生產程式的時候，你一般不會碰到這樣的問題，web服務、Android程式或者JavaFX將保持程式的存活。 Observable.interval()返回的是一個hot還是cold Observable？因為它是事件驅動的（頁數無限的），你可能會說它是hot。 但是，增加一個Observer，等5秒鐘，再加一個Observer。發生了什麼？請看：

import io.reactivex.Observable;
import java.util.concurrent.TimeUnit;

public class Launcher {
    public static void main(String[] args) {
        Observable<Long> seconds = Observable.interval(1, TimeUnit.SECONDS);
        //Observer 1
        seconds.subscribe(l -> System.out.println("Observer 1: " + l));
        //sleep 5 seconds
        sleep(5000);
        //Observer 2
        seconds.subscribe(l -> System.out.println("Observer 2: " + l));
        //sleep 5 seconds
        sleep(5000);        
    }

    public static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

輸出是這樣的：

Observer 1: 0
Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 0
Observer 1: 6
Observer 2: 1
Observer 1: 7
Observer 2: 2
Observer 1: 8
Observer 2: 3
Observer 1: 9
Observer 2: 4

5秒鐘過去了，Observer 2來了。注意，它有自己的計時器，從0開始。這兩個observers實際上有他們自己的emissions，每個都是從0開始。 所以，這個Observable實際上是cold。想讓所有的observers使用同一個計時器，可以使用ConnectableObservable：

import io.reactivex.Observable;
import java.util.concurrent.TimeUnit;

public class Launcher {
    public static void main(String[] args) {
        ConnectableObservable<Long> seconds =
                Observable.interval(1, TimeUnit.SECONDS).publish();
        //observer 1
        seconds.subscribe(l -> System.out.println("Observer 1: " + l));
        seconds.connect();
        //sleep 5 seconds
        sleep(5000);
        //observer 2
        seconds.subscribe(l -> System.out.println("Observer 2: " + l));
        //sleep 5 seconds
        sleep(5000);
    }

    public static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

輸出如下：

Observer 1: 0
Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 5
Observer 1: 6
Observer 2: 6
Observer 1: 7
Observer 2: 7
Observer 1: 8
Observer 2: 8
Observer 1: 9
Observer 2: 9

Observable.future()

RxJava的Observables比Futures更健壯和富有表現力，但是，如果你現在使用的庫仍然生產Futures，可以很容易地把他們轉換成Observables：

import io.reactivex.Observable;
import java.util.concurrent.Future;

public class Launcher {
    public static void main(String[] args) {
    Future<String> futureValue = ...;
    Observable.fromFuture(futureValue)
        .map(String::length)
        .subscribe(System.out::println);
    }
}

Observable.empty()

看上去好像沒什麼用，但是，Observable可以什麼都不發射，然後呼叫onComplete()：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
    Observable<String> empty = Observable.empty();
    
    empty.subscribe(System.out::println,
        Throwable::printStackTrace,
        () -> System.out.println("Done!"));
    }
}

空的observables通常代表空的資料集。有時候，有些運算也能返回空，比如filter()。有時候，你故意使用Observable.empty()。 empty Observable實質上就是RxJava的null。Empty Observables比null優雅，程式會繼續，不會拋NullPointerExceptions。

Observable.never()

Observable.empty()的近親是Observable.never()。唯一的不同是，它不呼叫onComplete()，observers一直在等emissions，卻永遠等不來：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
        Observable<String> empty = Observable.never();
        empty.subscribe(System.out::println,
                Throwable::printStackTrace,
                () -> System.out.println("Done!"));
        sleep(5000);
    }
}

這個Observable主要用來測試，生產中不經常使用。我們使用sleep()，就像Observable.interval()那樣，因為main執行緒產生Observable以後不會等待。 這個例子裡，我們等了5秒鐘，就是為了證明什麼都沒發射，然後，程式退出。

Observable.error()

你可以增加一個Observable，立刻呼叫onError()：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
        Observable.error(new Exception("Crash and burn!"))
                .subscribe(i -> System.out.println("RECEIVED: " + i),
                        Throwable::printStackTrace,
                        () -> System.out.println("Done!"));
    }
}

你也可以使用lambda提供異常，這樣給每個Observer提供了單獨的異常例項：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
        Observable.error(() -> new Exception("Crash and burn!"))
                .subscribe(i -> System.out.println("RECEIVED: " + i),
                        Throwable::printStackTrace,
                        () -> System.out.println("Done!"));
    }
}

Observable.defer()

Observable.defer()是一個強有力的工廠，為每個Observer增加單獨的狀態。當使用一定的Observable工廠的時候，如果你的源是有狀態的，你可能想給每個Observer增加單獨的狀態。你的源Observable可能不知道它的引數已經修改了，傳送過時的emissions。這裡有個簡單的例子，你有一個Observable.range()，有兩個靜態的int屬性：start和count。 如果你訂閱這個Observable，修改count，然後再次訂閱，你會發現，第二個Observer沒看到這個變化：

import io.reactivex.Observable;

public class Launcher {
    private static int start = 1;
    private static int count = 5;
    
    public static void main(String[] args) {
        Observable<Integer> source = Observable.range(start, count);
        source.subscribe(i -> System.out.println("Observer 1: " + i));
        //modify count
        count = 10;
        source.subscribe(i -> System.out.println("Observer 2: " + i));
    }
}

輸出是這樣的：

Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 1
Observer 2: 2
Observer 2: 3
Observer 2: 4
Observer 2: 5

想解決這個問題，你能玩兒每個訂閱者增加一個fresh Observable。使用Observable.defer()就可以實現，它接受一個lambda， 說明怎麼為每個訂閱者增加一個Observable。因為這樣做每次增加一個新的Observable，就可以反映引數的變化：

import io.reactivex.Observable;

public class Launcher {
    private static int start = 1;
    private static int count = 5;
    
    public static void main(String[] args) {
        Observable<Integer> source = Observable.defer(() ->
                Observable.range(start,count));
        source.subscribe(i -> System.out.println("Observer 1: " + i));
        //modify count
        count = 10;
        source.subscribe(i -> System.out.println("Observer 2: " + i));
    }
}

輸出是這樣的：

Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 1
Observer 2: 2
Observer 2: 3
Observer 2: 4
Observer 2: 5
Observer 2: 6
Observer 2: 7
Observer 2: 8
Observer 2: 9
Observer 2: 10

如果你的Observable源是原生實現，而且超過一個Observer（比如，重新使用一個Iterator，該迭代器只迭代一次），也可以使用Observable.defer()解決。

Observable.fromCallable()

如果你需要執行一個計算或者動作，然後發射它，你可以使用Observable.just()（或者Single.just()、Maybe.just()）。 但是，有時候，我們想晚一點再做，或者是deferred的方式。而且，如果處理過程發生錯誤，我們希望以onError()的方式發到Observable chain，而不是以Java傳統的方式拋異常。例如，如果你的Observable.just()想用1除以0,會拋異常，但是不發給Observer：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
    Observable.just(1 / 0)
            .subscribe(i -> System.out.println("RECEIVED: " + i),
                    e -> System.out.println("Error Captured: " + e));
    }
}

輸出是：

java.lang.ArithmeticException: / by zero

	at Launcher.main(Launcher.java:6)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)

如果希望響應式地處理錯誤，把錯誤發給Observer處理。上面的例子，可以使用Observable.fromCallable()，它接受一個lambda Supplier，發生錯誤時會發射給Observer：

import io.reactivex.Observable;

public class Launcher {
    public static void main(String[] args) {
        Observable.fromCallable(() -> 1 / 0)
            .subscribe(i -> System.out.println("Received: " + i),
                e -> System.out.println("Error Captured: " + e));
    }
}

輸出是：

Error Captured: java.lang.ArithmeticException: / by zero

相關推薦

RxJava學習 - 4. Other Observable sources

RxJava學習 - 4. Other Observable sources Observable.range() Observable.interval() Observable.future() Observable.empty() Obse

4. Other Observable sources

Observable.range() 可以使用Observable.range()發射一定範圍內的整數。從start值開始傳送發射，每次加1,直到一定的count。這些數通過onNext()事件傳遞，跟著一個onComplete()事件： import io.

Use WebP images along with other fallback sources to and placeholder to get max image optimization…

Use WebP images along with other fallback sources and placeholder to get max image optimization on the websiteI was figuring out the best way to AUTO optim

vs2017 +opencv 3.4.1 connected to each other

vs 與 opencv 的安裝不多說，百度一搜兩個軟體就可以官網獲取最新免費版本。 vs儘量不要安裝在C盤。 重點說以一下兩個軟體的適配： 首先，在安裝軟體時 ，不要急躁，否則，blabla~~~~ 步驟： (1) 更改環境變數 右擊 計算機  》》    屬性

JavaWeb網上圖書商城完整項目--day02-4.regist頁面提交表單時對所有輸入框進行校驗

word except 繼承 stub jstl use cti bmi imp 1、現在我們要將table表中的輸入的參數全部提交到後臺進行校驗，我們提交我們是按照表單的形式提交，所以我們首先需要在table表外面添加一個表單 <%@ page lang

第4次作業類測試代碼+105032014045+楊銘河

rfi color too efi rgs text blog ace val 1、類圖： 2、代碼： （1）計算類： class Arithmetic{　　//邏輯計算類 private int headphoneNum; private int

Robot Operating System (ROS)學習筆記4---語音控制

sla 語音 出現 tput http 學習 process 輸入 ubun 搭建環境：XMWare Ubuntu14.04 ROS（indigo） 轉載自古月居 轉載連接：http://www.guyuehome.com/260 一、語音識別包 1、安裝

第4次作業類測試代碼+019+李悅洲

stack app static jlabel field ted temp 函數 private 類圖： 代碼： package swingDesign; import java.awt.EventQueue; import javax.swing.JFra

2017.5.3 4.全排列

col font ace urn color span 順序輸出 所有 一個 題目描述 給定N(N<10)，按照字典序輸出所有的N排列。 輸入 第一行輸入N。 樣例輸入 3 輸出 輸出1到N的全排列，一行一個排列，按照字典序順序輸出。

修復Extjs5.1.4表格設置enableTextSelection: true之後，文本仍然不能選擇的BUG

over del ble nav targe find ret lec parent 如果您發現其他版本也有此BUG，可參照此方式進行修復，源代碼中多了一句攔截mousedown事件的代碼mousedownEvent.preventDefault()造成的。 Ext.def

10.4 補充範圍內丟失的值

trac emp sce coalesce min count int -1 left join select y.yr.coalesce(x.cnt,0) as cntfrom (select min_year-mod(cast(min_year as int),10)+

5.4下午

關於 更改 wid 對話框 提示 打開 span 下拉 詳細信息 關於Windows Server 2012的安裝在此不再贅述，同WIN8CP沒什麽差別，此文介紹一下我們在企業中部署的一項重要服務：活動目錄服務在新版服務器系統中的部署操作，也是屬於Step by Step

在Sql中將 varchar 值 '1,2,3,4,5,6' 轉換成數據類型 int

給定 序列 顯示 結果 空格 sel -方法 一個表 affect --問題：將aa轉換為Int類型失敗 string aa="3,5,11,56,88,45,23"; select * from ERPBuMen where ID in(aa) ; --方法sel

python-趣味百題4

[] esc 真分數 ins 個數字 nbsp res print turn 1.賽場統分在編程競賽中，有10個評委為參賽選手打分，分數為0-100分。選手最後得分為：去掉一個最高分和一個最低分後其余8個分數的平均值。思路：1）.隨機生成n個分數，放到列表s中2）.找到最高

2017.5.4下午學習內容

1-1 image .cn 英語 mage http 內容 ima 單詞 總結考研英語閱讀，並聽專業相關詞匯，復習上午所學單詞2017.5.4下午學習內容

5.4上午

區別 用戶 同時 src cnblogs oot roo 強制 tmp 文件系統類型： Linux中默認使用的文件系統類型是EXT3（EXT4），SWAP，同時也支持其他文件系統類型，如FAT16、FAT32、NTFS、XFS等。 目錄結構： 最頂

5.4下午 外教課 閱讀

ges home sftp cal ont star 童鞋 restart 建立 保存重啟,/etc/init.d/vsftpd restart 即可使用虛擬用戶登錄，這時候所有的虛擬用戶共同使用/home/ftpuser目錄上傳下載，如果想使用自己獨

高性能mysql 4,5,6章優化總結

重復 自適應 數據存儲 復雜 insert reat order 是我 條目 針對數據庫的優化，我們不能單純的說從哪一個方面，需要結合數據表的建立，數據類型的選擇，索引的設計和sql語句來考慮，我就針對怎麽建表，怎麽選擇數據類型，如何應用B-tree索引，hash索引和覆蓋

2017-5-4 線程

task void length tex bject button ring args strong 線程：幫助老板幹活的，不會幹擾老板的正常工作 如果一段代碼的執行需要時間，那麽必須開啟一個新線程來執行，如果不開線程，窗口會出現假死 開線程： Thread th = n

hadoop2.4 支持snappy

back war .com 適合 存儲空間 問題 borde pyc 道理 我們hadoop2,4集群默認不支持snappy壓縮，可是近期有業務方說他們的部分數據是snappy壓縮的（這部分數據由另外一個集群提供給他們時就是snappy壓縮格式的）想遷移到