RxJava詳しく解説_01【RxJavaとは】

このチュートリアルは RxJava1 拡張機能に基づいています。つまり、一方が情報を送信し、もう一方がその情報に応答して処理するコア フレームワーク コードを通じて行われます。

このフレームワークは、Microsoft アーキテクトの Erik Meijer 率いるチームによって開発され、2012 年 11 月にオープンソース化されました。

• Rx ライブラリは .NET、JavaScript、C++ などをサポートしており、現在ではほぼすべての一般的なプログラミング言語をサポートしています。

• Rx の言語ライブラリのほとんどは ReactiveX 組織によって保守されており、より一般的なものは RxJava/RxJS/Rx.NET であり、コミュニティ Web サイトは reactivex.io です。

• RxJava は人気のあるフレームワークであり、そのソース コードは GitHub に基づいています。RxJava のサポートに加えて、RxAndroid

• 2 のコードを簡素化するサポート フレームワークもあります。 Android プロジェクト、必要に応じて、バックグラウンドからデータを取得し、インターフェイスを更新します。コードは次のとおりです。例を見てみましょう:

  new Thread() {
      @Override
      public void run() {
          super.run();
          for (File folder : folders) {
              File[] files = folder.listFiles();
              for (File file : files) {
                  if (file.getName().endsWith(".png")) {
                      final Bitmap bitmap = getBitmapFromFile(file);
                      getActivity().runOnUiThread(new Runnable() {
                          @Override
                          public void run() {
                              imageCollectorView.addImage(bitmap);
                          }
                      });
                  }
              }
          }
      }
  }.start();

複数の層をネストすると、上記のコードは可読性が低すぎます。 RxJava を使用する場合は、次のように記述できます:

• Observable.from(folders)
      .flatMap(new Func1<File, Observable<File>>() {
          @Override
          public Observable<File> call(File file) {
              return Observable.from(file.listFiles());
          }
      })
      .filter(new Func1<File, Boolean>() {
          @Override
          public Boolean call(File file) {
              return file.getName().endsWith(".png");
          }
      })
      .map(new Func1<File, Bitmap>() {
          @Override
          public Bitmap call(File file) {
              return getBitmapFromFile(file);
          }
      })
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<Bitmap>() {
          @Override
          public void call(Bitmap bitmap) {
              imageCollectorView.addImage(bitmap);
          }
      });

  この方法で記述する利点は、コードを簡素化することに加えて、ネストのレベルを減らし、コードの読みやすさを向上させることです。各メソッドのスレッド。

3. フレームワークの紹介

現在、RxJava はバージョン 2.0 にアップグレードされていますが、RxJava をより深く理解するには、バージョン 1.0 から学習を始めることができます。 Android プロジェクトで RxJava をより適切に使用できるようにするために、gradle スクリプトの依存関係をプロジェクトに導入できます:

compile 'io.reactivex:rxandroid:1.2.1'
compile 'io.reactivex:rxjava:1.1.6'

これで、私たちのプロジェクトはすでに RxJava の機能をサポートしています。

4. 応答性の核心

いわゆる応答性は 2 つの部分の存在にすぎません。1 つの部分はイベント/メッセージの送信を担当し、もう 1 つの部分はイベント/メッセージへの応答を担当します。

昔は、ニュースを読みたければ、通常、新聞を読まなければなりませんでした。たとえば、特定の新聞や雑誌に興味がある場合、最初に 3 つのことを行う必要があります:

自宅の住所を入力する

該当する新聞を見つける

新聞にアクセスして購読する1ヶ月分の新聞
上記のプロセスを経て、毎日新しい新聞や定期刊行物が出るたびに、新聞社からあなたのご自宅に雑誌が届きます。

3. 上記の例のコード抽象化、手順は次のとおりです:

オブザーバーを提供します (あなたは雑誌の内容に関心がある人なので、イベントを観察している人です)

監視対象のオブザーバーを提供します（新しい雑誌が出る限り、興味のある人に通知する必要があるため、新聞社が監視の対象になります）

定期購読（つまり、オブザーバーと監視対象監視されているオブジェクトが変化したときに、イベントを監視しているオブジェクトに即座に通知されるように、相互に関連付ける必要があります)

2. 上記の例のデモ コードは次のとおりです:

//1.创建被观察者
Observable<String> observable = 
        Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //4.开始发送事件 
                //事件有3个类型 分别是onNext() onCompleted() onError()
                //onCompleted() onError() 一般都是用来通知观察者 事件发送完毕了,两者只取其一。
                subscriber.onNext("Hello Android !");
                subscriber.onNext("Hello Java !");
                subscriber.onNext("Hello C !");
                subscriber.onCompleted();
            }
        });

//2.创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onCompleted() {
        Log.i(TAG, "onCompleted ");
    }

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

    @Override
    public void onNext(String s) {
        Log.i(TAG, "onNext: "+s);
    }
};

//3.订阅
observable.subscribe(subscriber);

3. 出力は次のとおりです。

   com.m520it.rxjava I/IT520: onNext: Hello Android !
   com.m520it.rxjava I/IT520: onNext: Hello Java !
   com.m520it.rxjava I/IT520: onNext: Hello C !
   com.m520it.rxjava I/IT520: onCompleted

   コードの実行方法

上記のコードでは、オブザーバーのサブスクライバーがサブスクライブすると、オブザーバーが監視可能になった後、システムは監視可能なオブジェクト内の call() を自動的にコールバックします。

onNext/onCompleted/onError などのイベントを送信した後、observable の call() メソッド エンティティ内。

その後、サブスクライバーは対応するメソッドにコールバックできます。
5. Observable のバリアント

• Ordinary Observable の送信には onNext、onError、onCompleted の 3 つのメソッドが必要ですが、Single は Observable のバリアントとして 2 つのメソッドのみを必要とします:

• - Single はこれに 1 つの値を送信します。 Method

onError - 必要な値を発行できない場合、Single はこのメソッドに Throwable オブジェクトを発行します

Single はこれら 2 つのメソッドのいずれか 1 つだけを呼び出します。どちらがメソッドであるかに関係なく、一度だけ呼び出されます。このメソッドの後、サブスクリプション関係は終了します。

• final Single<String> single = Single.create(new Single.OnSubscribe<String>() {
              @Override
              public void call(SingleSubscriber<? super String> singleSubscriber) {
                  //先调用onNext() 最后调用onCompleted() 
                  //singleSubscriber.onSuccess("Hello Android !");
                  //只调用onError();
                  singleSubscriber.onError(new NullPointerException("mock Exception !"));
              }
          });
  
  Observer<String> observer = new Observer<String>() {
      @Override
      public void onCompleted() {
          Log.i(TAG, "onCompleted ");
      }
  
      @Override
      public void onError(Throwable e) {
          Log.i(TAG, "onError: "+e.getLocalizedMessage());
      }
  
      @Override
      public void onNext(String s) {
          Log.i(TAG, "onNext: "+s);
      }
  };
  single.subscribe(observer);

  6. Observer のバリアント

Observer オブザーバー オブジェクト。上記では代わりに Subscriber オブジェクトを使用します。オブジェクト自体が Observer を継承しているためです。

• このオブジェクトは onNext()&onCompleted()&onError() イベントを実装します。どのイベントを重視する場合は、対応するメソッドを実装するだけで済みます。

  //创建观察者
  Subscriber<String> subscriber = new Subscriber<String>() {
      @Override
      public void onCompleted() {
          Log.i(TAG, "onCompleted ");
      }
  
      @Override
      public void onError(Throwable e) {
          Log.i(TAG, "onError: "+e.getLocalizedMessage());
      }
  
      @Override
      public void onNext(String s) {
          Log.i(TAG, "onNext: "+s);
      }
  };
  
  //订阅
  observable.subscribe(subscriber);

上記のコードでは、 onNext () イベントのみを考慮しますが、onCompleted()&onError() イベントを実装する必要があります。このようなコードは非常に肥大化しているように見えます。この需要を考慮して、RxJava フレームワークはサブスクリプションに特別な調整を加えました。コードは次のとおりです:

//为指定的onNext事件创建独立的接口
Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        Log.i(TAG, "call: "+s);
    }
};

//订阅
observable.subscribe(onNextAction);

気づいたかどうかはわかりませんが、subscribe() はオブザーバーではなく、特定の onNext インターフェイスをサブスクライブします。物体。同様の関数は次のとおりで、必要に応じて対応するサブスクリプションを実装できます:

public Subscription subscribe(final Observer オブザーバー)

public Subscription subscribe(final Action1 onNext)

public Subscription subscribe(final Action1 onNext, Action1 onError)

public Subscription subscribe(final Action1 onNext, Action1 onError, Action0 onCompleted)

这里还有一个forEach函数有类似的功能：

• public void forEach(final Action1 onNext)

• public void forEach(final Action1 onNext, Action1 onError)

• public void forEach(final Action1 onNext, Action1 onError, Action0 onComplete)

7.Subject变种

上面2节中既介绍了被观察者变种，又介绍了观察者变种，这里再介绍一种雌雄同体的对象(既作为被观察者使用，也可以作为观察者)。

针对不同的场景一共有四种类型的Subject。他们并不是在所有的实现中全部都存在。

AsyncSubject

一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。它会把这最后一个值发射给任何后续的观察者。

以下贴出代码：

//创建被观察者
final AsyncSubject<String> subject = AsyncSubject.create();
//创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onCompleted() {
        Log.i(TAG, "onCompleted");
    }

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

    @Override
    public void onNext(String s) {
        Log.i(TAG, "s:" + s);

    }
};
//订阅事件
subject.subscribe(subscriber);
//被观察者发出事件 如果调用onCompleted(),onNext()则会打印最后一个事件；如果没有,onNext()则不打印任何事件。
subject.onNext("Hello Android ");
subject.onNext("Hello Java ");
subject.onCompleted();

输出：

s:Hello Java 
onCompleted

然而，如果原始的Observable因为发生了错误而终止，AsyncSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

上面的观察者被观察者代码相同,现在发出一系列信号，并在最后发出异常 代码如下：

subject.onNext("Hello Android ");
subject.onNext("Hello Java ");
//因为发送了异常 所以onNext()无法被打印
subject.onError(null);

BehaviorSubject

当观察者订阅BehaviorSubject时，他会将订阅前最后一次发送的事件和订阅后的所有发送事件都打印出来，如果订阅前无发送事件，则会默认接收构造器create(T)里面的对象和订阅后的所有事件，代码如下：

BehaviorSubject subject=BehaviorSubject.create("NROMAL");

Subscriber subscriber = new Subscriber() {
    @Override
    public void onCompleted() {
        Log.i(TAG, "onCompleted");
    }

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

    @Override
    public void onNext(Object o) {
        Log.i(TAG, "onNext: " + o);
    }
};

//subject.onNext("Hello Android !");
//subject.onNext("Hello Java !");
//subject.onNext("Hello C !");
//这里开始订阅 如果上面的3个注释没去掉，则Hello C的事件和订阅后面的事件生效
//如果上面的三个注释去掉 则打印构造器NORMAL事件生效后和订阅后面的事件生效
subject.subscribe(subscriber);

subject.onNext("Hello CPP !");
subject.onNext("Hello IOS !");

PublishSubject

PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。

需要注意的是，PublishSubject可能会一创建完成就立刻开始发射数据，因此这里有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。

代码如下：

PublishSubject subject= PublishSubject.create();

Action1<String> onNextAction1 = new Action1<String>(){

    @Override
    public void call(String s) {
        Log.i(TAG, "onNextAction1 call: "+s);
    }
};

Action1<String> onNextAction2 = new Action1<String>(){

    @Override
    public void call(String s) {
        Log.i(TAG, "onNextAction2 call: "+s);
    }
};

subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");

输出如下：

onNextAction1 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !

ReplaySubject

ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。

代码如下：

ReplaySubject subject= ReplaySubject.create();

Action1<String> onNextAction1 = new Action1<String>(){

    @Override
    public void call(String s) {
        Log.i(TAG, "onNextAction1 call: "+s);
    }
};

Action1<String> onNextAction2 = new Action1<String>(){

    @Override
    public void call(String s) {
        Log.i(TAG, "onNextAction2 call: "+s);
    }
};

subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");

输出如下：

onNextAction1 call: Hello Android !
onNextAction1 call: Hello Java !
onNextAction2 call: Hello Android !
onNextAction2 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !

Subject总结

• AsyncSubject无论何时订阅 只会接收最后一次onNext()事件，如果最后出现异常，则不会打印任何onNext()

• BehaviorSubject会从订阅前最后一次oNext()开始打印直至结束。如果订阅前无调用onNext(),则调用默认creat(T)传入的对象。如果异常后才调用，则不打印onNext()

• PublishSubject只会打印订阅后的任何事件。

• ReplaySubject无论订阅在何时都会调用发送的事件。

 以上就是深入浅出RxJava_01[什么是RxJava] 的详细介绍的内容，更多相关内容请关注PHP中文网（www.php.cn）！