2. 程式人生 > >RX之非同步操作（start、toAsync、startFuture、deferFuture、fromAction、fromRunnable、forEachFuture、runAsync）

RX之非同步操作（start、toAsync、startFuture、deferFuture、fromAction、fromRunnable、forEachFuture、runAsync）

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

需要在buld.gradle中加入compile 'io.reactivex:rxjava-async-util:0.21.0'

一、start

返回一個Observable，它發射一個類似於函式宣告的值。程式語言有很多種方法可以從運算結果中獲取值，它們的名字一般叫functions, futures, actions, callables, runnables等等。在Start目錄下的這組操作符可以讓它們表現得像Observable，因此它們可以在Observables呼叫鏈中與其它Observable搭配使用。Start操作符的多種RxJava實現都屬於可選的rxjava-async模組。

rxjava-async

模組包含start操作符，它接受一個函式作為引數，呼叫這個函式獲取一個值，然後返回一個會發射這個值給後續觀察者的Observable。

注意：這個函式只會被執行一次，即使多個觀察者訂閱這個返回的Observable。

 Observable<Integer> observable = Async.start(new Func0<Integer>() {
            @Override
            public Integer call() {
                //函式內為非同步操作
                try {
                    Thread.sleep(5 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return 20;
            }
        });

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer v) {
                Log.e(TAG, "onNext................." + v);

            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError.....................");
            }
        };

        observable.subscribe(subscriber);
5s後才輸出列印結果
執行結果：


二、toAsync

rxjava-async模組還包含這幾個操作符：toAsync, asyncAction, 和asyncFunc。它們接受一個函式或一個Action作為引數。對於函式(functions)，這個操作符呼叫這個函式獲取一個值，然後返回一個會發射這個值給後續觀察者的Observable（和start一樣）。對於動作(Action)，過程類似，但是沒有返回值，在這種情況下，這個操作符在終止前會發射一個null值。

注：這個函式或動作只會被執行一次，即使多個觀察者訂閱這個返回的Observable。

 Func0<Observable<Integer>> func0 = Async.toAsync(new Func0<Integer>() {
            @Override
            public Integer call() {
                //函式內為非同步操作
                try {
                    Thread.sleep(5 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return 20;
            }
        });
        Observable<Integer> observable = func0.call();

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer v) {
                Log.e(TAG, "onNext................." + v);

            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError.....................");
            }
        };
        observable.subscribe(subscriber);

執行結果：


三、startFuture

xjava-async模組還包含一個startFuture操作符，傳遞給它一個返回Future的函式，startFuture會立即呼叫這個函式獲取Future物件，然後呼叫Future的get()方法嘗試獲取它的值。它返回一個發射這個值給後續觀察者的Observable。

final Future<Integer> future = Executors.newCachedThreadPool().submit(new Task());
        Observable<Integer> observable = Async.startFuture(new Func0<Future<Integer>>() {
            @Override
            public Future<Integer> call() {
                //函式內為非同步操作
                return future;
            }
        });


        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer v) {
                Log.e(TAG, "onNext................." + v);

            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError.....................");
            }
        };

        observable.subscribe(subscriber);

執行結果：

10s後才輸出列印結果


四、deferFuture

deferFuture中的函式可以非同步執行一些操作，當完成後返回一個Observable，但是這個Observable不會立刻發射資料，直到開始訂閱時。 

rxjava-async模組還包含一個deferFuture操作符，傳遞給它一個返回Future的函式（這個Future返回一個Observable），deferFuture返回一個Observable，但是不會呼叫你提供的函式，知道有觀察者訂閱它返回的Observable。這時，它立即呼叫Future的get()方法，然後映象發射get()方法返回的Observable發射的資料。

用這種方法，你可以在Observables呼叫鏈中包含一個返回Observable的Future物件。

 Observable<Long> observable = Observable.interval(1,TimeUnit.SECONDS).take(5);

        final Future<Observable<Integer>> future = Executors.newCachedThreadPool().submit(new ObservableTask());
        final Observable<Integer> observable1 = Async.deferFuture(new Func0<Future<Observable<Integer>>>() {
            @Override
            public Future<Observable<Integer>> call() {
                return future;
            }
        });

        final Subscriber<Integer> subscriber1 = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext1................." + integer);
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted1.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError1.....................");
            }
        };

        Subscriber<Long> subscriber = new Subscriber<Long>() {
            @Override
            public void onNext(Long integer) {
                Log.e(TAG, "onNext................." + integer);
                if(integer == 4){
                    observable1.subscribe(subscriber1);
                }
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError.....................");
            }
        };


        observable.subscribe(subscriber);

執行結果：


五、fromAction

當fromAction()中的函式Action0執行完成後發射資料。Action無返回結果。rxjava-async模組還包含一個fromAction操作符，它接受一個Action作為引數，返回一個Observable，一旦Action終止，它發射這個你傳遞給fromAction的資料。

        final int result = 1;
        final Observable<Integer> observable = Async.fromAction(new Action0() {
            @Override
            public void call() {

                for(int i=1;i<10;i++){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            }
        },result);

        final Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext1................." + integer);
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted1.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError1.....................");
            }
        };

        observable.subscribe(subscriber);
執行結果：


六、fromCallable

當fromCallable中的函式執行完成後將其結果發射出去。rxjava-async模組還包含一個fromCallable操作符，它接受一個Callable作為引數，返回一個發射這個Callable的結果的Observable。

 final Observable<Integer> observable = Async.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int result = 1;
                for(int i=1;i<10;i++){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    result *= i;
                }
                return result;
            }
        });

        final Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext1................." + integer);
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted1.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError1.....................");
            }
        };

        observable.subscribe(subscriber);
執行結果：


七、fromRunnable

當run方法執行完成後傳送傳入的引數資料。rxjava-async模組還包含一個fromRunnable操作符，它接受一個Runnable作為引數，返回一個Observable，一旦Runnable終止，它發射這個你傳遞給fromRunnable的資料。

        //當runnable執行完成後所需發射的資料
        int result = 20;
         Observable<Integer> observable = Async.fromRunnable(new Runnable() {
             @Override
             public void run() {
                 for(int i=1;i<10;i++){
                     try {
                         Thread.sleep(1000);
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                     }
                 }
             }
         },result);

         Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext1................." + integer);
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted1.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError1.....................");
            }
        };

        observable.subscribe(subscriber);
執行結果：


八、forEachFuture

rxjava-async模組還包含一個forEachFuture操作符。它其實不算Start操作符的一個變體，而是有一些自己的特點。你傳遞一些典型的觀察者方法（如onNext, onError和onCompleted）給它，Observable會以通常的方式呼叫它。但是forEachFuture自己返回一個Future並且在get()方法處阻塞，直到原始Observable執行完成，然後它返回，完成還是錯誤依賴於原始Observable是完成還是錯誤。

如果你想要一個函式阻塞直到Observable執行完成，可以使用這個操作符。


private void testForEachFuture() {

        Async.forEachFuture(Observable.just(1, 2, 3, 4, 5),
                new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        newInfo(integer);

                    }
                });

    }


    private void newInfo(Integer integer){
        Log.e(TAG, "newInfo.................收到的資料是：" + integer+"..............轉換成："+integer*20);
    }
執行結果：



九、runAsync

任何時候都可以通過subscription的unsubscrip來停止訂閱。rxjava-async模組還包含一個runAsync操作符。它很特殊，返回一個叫做StoppableObservable的特殊Observable。傳遞一個Action和一個Scheduler給runAsync，它返回一個使用這個Action產生資料的StoppableObservable。這個Action接受一個Observable和一個Subscription作為引數，它使用Subscription檢查unsubscribed條件，一旦發現條件為真就立即停止發射資料。在任何時候你都可以使用unsubscribe方法手動停止一個StoppableObservable（這會同時取消訂閱與這個StoppableObservable關聯的Subscription）。

由於runAsync會立即呼叫Action並開始發射資料，在你建立StoppableObservable之後到你的觀察者準備好接受資料之前這段時間裡，可能會有一部分資料會丟失。如果這不符合你的要求，可以使用runAsync的一個變體，它也接受一個Subject引數，傳遞一個ReplaySubject給它，你可以獲取其它丟失的資料了。在RxJava中還有一個版本的From操作符可以將Future轉換為Observable，與start相似。

 //action所需要執行的執行緒
        Scheduler scheduler = Schedulers.newThread();
        Action2<? super Observer<? super Integer>, ? super Subscription> action = new Action2<Observer<? super Integer>, Subscription>() {
            @Override
            public void call(Observer<? super Integer> observer, Subscription subscription) {
               int i = 1;
                while (!subscription.isUnsubscribed()){
                    Log.e(TAG, "call................observer.onNext");
                    observer.onNext(i);
                    i += 1;
                    if(i == 4){
                        subscription.unsubscribe();
                        observer.onCompleted();
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            }
        };

       StoppableObservable<Integer> stoppableObservable  =  Async.runAsync(scheduler, action);

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext................." + integer);
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted.................");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError.....................");
            }
        };

        stoppableObservable.subscribe(subscriber);

執行結果：


相關推薦

RX之非同步操作（start、toAsync、startFuture、deferFuture、fromAction、fromRunnable、forEachFuture、runAsync）

需要在buld.gradle中加入compile 'io.reactivex:rxjava-async-util:0.21.0' 一、start 返回一個Observable，它發射一個類似於函式宣告的值。程式語言有很多種方法可以從運算結果中獲取值，它們的名字一般叫func

sqlite3學習之資料庫操作（建立、附加、分離）

SQLite 的 sqlite3 命令被用來建立新的 SQLite 資料庫，所以我們可以不需要任何特殊的許可權即可建立一個數據庫，語法格式也很簡單，如下： sqlite3 DatabaseName.db 執行完上述命令之後，會在你的當前目錄下生成一個你的資料庫名稱加上.db做字尾的檔案，

sqlite3學習之表操作（建立、刪除）

SQLite 的 CREATE TABLE 語句用於在任何給定的資料庫建立一個新表。建立基本表，涉及到命名錶、定義列及每一列的資料型別，語法很簡單，如下： CREATE TABLE database_name.table_name( column1 datatype PRIMARY

python selenium 元素操作之鍵盤操作（五）

兩個 .com multi 模擬 刪除 del escape comm f11 上節介紹了模擬鼠標對元素的操作，本節主要介紹鍵盤對元素的操作，實際過程中鼠標對元素的操作比鍵盤對元素的操作更經常使用，但是鍵盤對元素的操作也很重要，本節主要介紹一下鍵盤對元素的操作。 selen

vue.js之常操作（實例）

tex styles title oct lar per click 一個 lis 聽說大家都開始用react、vue這些MVVM框架了，為了不落後，還是學學吧！（之前只對angular了解一點，時間一長，也忘得差不多了，所以學習vue相當於從小白開始） 從vue.js官網

docker 之 基本操作 （幹貨）

eight -s 本地 arch 進入容器 name height 官方 -- 1、下載鏡像：docker pull <name>2、查看本地鏡像：docker images3、查看官方鏡像：docker search <name>4、運行容器：d

Spring之事務操作（配置文件）

5.x ati ole 操作 print use point run div UserDao.java 1 package helloworld.tx; 2 3 import org.springframework.jdbc.core.JdbcTemplate;

Java之JSON操作（Jackson）

單個 string not chan writev except java對象 tom ora Java to JSON： 1 package json.jackson; 2 3 import bean.User; 4 import com.fasterxml.

tensorflow 佇列管理之非同步操作

# -*- coding: utf-8 -*- """ Created on Thu Oct 25 09:49:34 2018 @author: Grey """ import tensorflow as tf #模擬非同步子執行緒存入樣本，主執行緒讀取樣本 #1.定義一個佇列，100 Q =

Java基礎之IO操作（二）

Java基礎之IO操作（二） 一、BufferedInputStream public static void main(String[] args) { File src = new File("abc.txt"); //2、選擇流 InputStream is =null;

Java基礎之IO操作（一）

Java基礎之IO操作（一） 一、FileInputStream 第一個程式，理解操作步驟 public static void main(String[] args) { //1、建立源 File src = new File("abc.txt"); //2、

SpringBoot2.0之非同步任務（async）和 java8 CompletableFuture的呼叫

1.我們先介紹java8 的非同步呼叫：CompletableFuture 用CompletableFuture.supplyAsync（）定義要執行的非同步任務 public class java8Async { public static void main(Str

python學習筆記之基礎操作（五）字串格式化（1）%格式化

格式化基本格式 s = "i am %s ,i am %d years old ,and i like %s"%("haiyang",20,"zhu") print(s) #區分資料型別，無法自動在字串

python學習筆記之基礎操作（五）字串格式化（2）format

format格式化 1,基本格式 #對於每一個大括號，在後面的引數中找到對應的引數插進來 #format操作類似於於將傳入的引數製成多個數據的資料結構元組或者字典，然後依照索引插入引數 s = "i a

MongoDB 學習之基本操作（五）

基本操作 MongoDB將資料儲存為一個文件，資料結構由鍵值(key=>value)對組成 MongoDB文件類似於JSON物件，欄位值可以包含其他文件、陣列、文件陣列 安裝管理mongod

【Springboot】之 非同步處理（@Async）

前言 使用非同步是為了縮短等待時間。 即：主執行緒儘早處理完並返回資訊，能更快響應；而副執行緒執行其他操作用於完善。 一、配置類 作用：配置執行緒池，實現執行緒複用。 通過 @EnableAsy

MongoDB 學習之高階操作（七）

高階操作 講解關於mongodb的高階操作，包括聚合、主從複製、分片、備份與恢復、MR 完成python與mongodb的互動 聚合 aggregate 聚合(aggregate)主要用於計算資料

python學習筆記之基礎操作（六）生成器和迭代器

生成器總結 語法上和函式類似：生成器函式幾乎和常規函式一樣，不同的又兩點， 第一點是返回使用yield關鍵字，第二點是在連續邏輯內可以多次返回 這是yield的特點決定的，yield會記住函式執行的位

spring之AOP操作（基於aspectJ實現）--配置檔案和註解兩種方式實現

AOP概念 　　1 aop：面向切面（方面）程式設計，擴充套件功能不修改原始碼實現   　　2  AOP採取橫向抽取機制，取代了傳統縱向繼承體系重複性程式碼   　　3 aop底層使用動態代理實現 　　　　（1）第一種情況，有介面情況，使用動態代理建立介面實現類代理物

JS中的非同步操作（轉）

JS中非同步程式設計的方法有： 回撥函式 事件監聽 釋出/訂閱 promise generator（ES6） async/await(ES7) 回撥函式 回撥是非同步程式設計中最基礎的方法。舉例一個簡單的回撥：在f1執行完之後再執行f2 var func1