[Android] RxJava の最初の章

RxJava について

RxJava は、Java VM 環境で使用するために ReactiveX によって起動された非同期操作ライブラリです。 Java 環境に加えて、ReactiveX は、Py、Js、Go などの他のプログラミング言語用の Rx ライブラリも起動します。インターネット上には RxJava の紹介や使い方が数多くあり、Android 開発でも RxJava を使用したプロジェクトが数多くあります。では、なぜ RxJava を使用するのでしょうか? Android 開発では、開発者が使用できる非同期操作メソッドも提供されています。RxJava は、Handle や AsyncTask よりもシンプルで洗練されていると思います。

1 RxJava は、プログラム ロジックが明確かつ簡潔であるチェーン コールを使用します。

2 拡張オブザーバー デザイン パターンを採用します。

オブザーバー パターンとその他の RxJava の紹介は、主に繰り返しません。 RxJava と RxAndroid 。 RxJava の公式ドキュメントを詳しく紹介しています。このセクションは主に学習と議論を目的としています。間違いがある場合はご指摘いただければ幸いです。

Observed Observable

RxJava を使用して、データの出力に使用される Observable を作成する必要があります。次の Observable の create メソッドは、Action1 を継承する OnSubscribe を渡す必要があります。アクションの Subscriber はサブスクライバです。

public static <T> Observable<T> create(OnSubscribe<T> f) { 
    return new Observable<T>(RxJavaHooks.onCreate(f)); 
}

さらに、create メソッドはインターフェイス呼び出しを実装し、サブスクライバー オブジェクトを返す必要があります。 call メソッドは、オブザーバブルがサブスクライブされた後に実行されるイベント ストリームを実装します。 subscriber.onNext はデータを出力し、subscriber.onCompleted は出力イベントの終了を示すことができます。次に、オブザーバブルの submit メソッドを呼び出して、サブスクライブ後に実行されるイベント ストリームを実装します。

Observable<String> observable = Observable 
                .create(new Observable.OnSubscribe<String>() { 
 
            @Override 
            public void call(Subscriber<? super String> subscriber) { 
                subscriber.onNext("1"); 
                subscriber.onNext("2"); 
                subscriber.onNext("3"); 
                subscriber.onNext("4"); 
                subscriber.onNext("5"); 
            } 
}); 
Subscriber<String> subscriber = new  Subscriber<String>() { 
            @Override 
            public void onCompleted() { 
 
            } 
 
            @Override 
            public void onError(Throwable e) { 
 
            } 
 
            @Override 
            public void onNext(String s) { 
                System.out.print(s + &#39;\n&#39;); 
            } 
}; 
observable.subscribe(subscriber); 
//输出结果 print： 
//1 
//2 
//3 
//4 
//5

create メソッドを使用して Observable を作成することに加えて、from または just を使用して、発行されたイベント ストリームをすばやくセットアップし、作成手順を簡素化することもできます。

Observable<String> o = Observable.from("a", "b", "c");

Observable<String> o = Observable.just("one object");

非同期操作

RxJava のスレッドは、特定の操作が実行されるスレッドを制御する Schedulers スケジューラによって制御されます。

Schedulers.immediate() 現在のスレッドで実行します

Schedulers.newThread() 実行するタスクごとにスレッドを作成します

Schedulers.computation() タスクが実行されるスレッドを計算します

Schedulers.io() IOタスクが実行されるスレッド....

AndroidSchedulers.mainThread() Androidのメインスレッドが実行されます

スレッド制御は主にsubscribeOn()とobservableの2つのメソッドによって制御されます:

subscribeOnはObservableのスレッドを制御します。 OnSubscribe が配置されます。これは、Create、just、from が配置される Observable スレッドと同等です。

observeOn は、サブスクライバーのスレッドを制御します。これは、制御イベントが実行されるスレッドとも言えます。

Observable 
        .just(1,2,3) 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Subscriber<Integer>() { 
            @Override 
            public void onCompleted() { 
 
            } 
 
            @Override 
            public void onError(Throwable e) { 
 
            } 
 
            @Override 
            public void onNext(Integer integer) { 
                 System.out.print(integer + &#39;\n&#39;);                        
            } 
}); 
//输出结果 print： 
//1 
//2 
//3

上記の RxJava チェーン呼び出しコードを作成しました。以前に使用した非同期呼び出しよりもはるかにクリーンだと思いますか? これは Virgos にとって非常に良いものだとも言いました。

Operators

ReactiveX はそれぞれ多くの演算子を提供します。 Operator にはさまざまな機能がありますが、その目的は、Observable と Subscribe の間で発行されたイベント ストリームを変換および変更することです。このセクションでは、いくつかの一般的で単純な演算子を紹介します。後で機会があれば、演算子に関する別のセクションを作成して、各演算子の役割を詳しく説明します。公式のオペレーター文書に添付されているので、その数がわかります。

Map()

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) { 
        return create(new OnSubscribeMap<T, R>(this, func)); 
    }

まず、演算子マップを導入して、T 型データを R 型データに変換し、R 型データを返す Func1 インターフェイスを実装します。たとえば、整数型のイベント キューは渡され、マップ処理後に文字列型として返されます。

Observable 
                .just(1,2,3) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                    ...... 
                    @Override 
                    public void onNext(String str) { 
                        System.out.print(str + &#39;\n&#39;); 
                    } 
                }); 
//输出结果 print： 
//1 
//2 
//3

Filter()

public final Observable<T> filter(Func1<? super T, Boolean> predicate) { 
       return create(new OnSubscribeFilter<T>(this, predicate)); 
   }

filter はマップと同様に Func1 インターフェイスを実装しますが、その変換された型はブール値であり、変換されたブール値が true の場合、サブスクライバーはそれを受信できます。フィルタリングを通過しない場合、イベントは消費されません。たとえば、イベント ストリーム フィルタリングでは、配信を続ける前に int 値が 2 で割り切れる必要があるため、サブスクライバが消費できる最終イベントは 2、4、6、8、および 10 になります。

Observable 
                .just(1,2,3,4,5,6,7,8,9,10) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .filter(new Func1<Integer, Boolean>() { 
                    @Override 
                    public Boolean call(Integer integer) { 
                        return integer % 2 == 0; 
                    } 
                }) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                   ...... 
                    @Override 
                    public void onNext(String str) { 
                        System.out.print(str + &#39;\n&#39;); 
                        Log.i("subscribe", str); 
                    } 
                }); 
//输出结果 print： 
//2 
//3 
//4 
//6 
//8 
//10

Skip()

public final Observable<T> skip(int count) { 
        return lift(new OperatorSkip<T>(count)); 
    }

スキップ演算子は、最初のいくつかのイベントをスキップし、特定のイベントからイベントの発行を開始することを意味し、添字は 0 から始まります。

Observable 
                .just(1,2,3,4,5,6,7,8,9,10) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .skip(3) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                    ...... 
                    @Override 
                    public void onNext(String s) { 
                        System.out.print(s + &#39;\n&#39;); 
                        Log.i("subscribe", s); 
                    } 
                }); 
//输出结果 print： 
//4 
//5 
//6 
//7 
//8 
//9 
//10

Range()

public static Observable<Integer> range(int start, int count) { 
        if (count < 0) { 
            throw new IllegalArgumentException("Count can not be negative"); 
        } 
        if (count == 0) { 
            return Observable.empty(); 
        } 
        if (start > Integer.MAX_VALUE - count + 1) { 
            throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE"); 
        } 
        if(count == 1) { 
            return Observable.just(start); 
        } 
        return Observable.create(new OnSubscribeRange(start, start + (count - 1))); 
    }

range 演算子は、単に、出力される連続 int 型配列を渡すものとして理解できます。n は開始 int 値、m は Count です。たとえば、n = 1、m = 5 の int 配列は {1, 2, 3, 4, 5} になります

End

私はこの部分を最初に学びました。これが、RxJava についての私の最初の理解と学習です。実際、RxJava の使用は主に演算子の使用に依存します。前に紹介した演算子は最も単純で基本的なものであり、まだ紹介されていない特に便利な演算子がたくさんあります。今後もいくつかの演算子を紹介していきます。 RxJava は、その能力の高さから Android 開発で非常に人気があります。同時に、RxJava を Retrofit と組み合わせて使用​​すると、ネットワーク リクエストの戻り値をより効率的に処理できます。さらに、GitHub GoogleSample の android-architecture にも RxJava フレームワークを使用した TODO プロジェクトがあり、プロジェクトでの RxJava の実際のアプリケーションを確認して理解することができます。