就可以应用上百个操作符来实现自己的表达式,首先会想到的是使用Observable

在大多数 RxJava 示例代码和教程中出现最为频繁的一个类 —— Observable,它是产生响应式编程魔力的关键。它的用法很简单,只需要跟踪 3 个事件 —— onNextonErroronCompleted就可以应用上百个操作符来实现自己的表达式。那么为什么你还需要了解其他东西?

Maybe tomorrow.jpeg

转载自:https://xiaobailong24.me/2017/03/18/Android-RxJava2.x/

但是你仔细思考下,你真的需要每次都知道这 3 个事件吗?实际上,在大多数情况下并不需要。ReactiveX 文档中讲述的基本都是关于连续的事件流,因此我们经常忘记通常我们关心的只是监听单一事件或者只监听 completed or failed 事件。

通常情况下,如果我们想要使用 RxJava 首先会想到的是使用Observable,如果要考虑到Backpressure的情况,在 RxJava2.x 时代我们会使用Flowable。除了Observable和Flowable之外,在 RxJava2.x 中还有三种类型的Observables:Single、Completable、Maybe。

前言
最近学习了一下RxJava,发现是个好东西,有点相见恨晚的感觉,一开始学习了RxJava 1.x,看了很多国内的博客,有点理解了,后来发现现在都 2.x 了,于是各种搜索,最后发现Season_zlc写的系列教程通俗易懂,非常适合初学者。因此在这里记录一下学习过程,感谢原作者的无私分享,希望在实战中提升对 RxJava 的理解。
Gradle 配置
要在Android中使用RxJava2, 先添加Gradle配置,最新的版本可在GitHub找到。RxJava、RxAndroid
1
2

在这种情况下我们应该考虑用 RxJava 的两个绝妙的设计 —— Single<T> 和 Completable,在分析两者之前,让我们先看看他们应用场景的示例。

类型 描述
Observable<T> 能够发射0或n个数据,并以成功或错误事件终止。
Flowable<T> 能够发射0或n个数据,并以成功或错误事件终止。 支持Backpressure,可以控制数据源发射的速度。
Single<T> 只发射单个数据或错误事件。
Completable 它从来不发射数据,只处理 onComplete 和 onError 事件。可以看成是Rx的Runnable。
Maybe<T> 能够发射0或者1个数据,要么成功,要么失败。有点类似于Optional

compile "io.reactivex.rxjava2:rxjava:2.0.7"
compile "io.reactivex.rxjava2:rxandroid:2.0.1"

本文中所有代码都是基于 RxJava 2.x ,不是 1.x 版本。如果你还没升级 RxJava 到最新的 2.x 版本, 强烈建议你马上升级。

从上面的表格可以看出,这五种被观察者类型中只有Flowable能支持Backpressure,如果有需要Backpressure的情况,还是必须要使用Flowable。

第一式 亢龙有悔
概述
网上也有很多介绍RxJava原理的文章,通常这些文章都从观察者模式开始,先讲观察者,被观察者,订阅关系巴拉巴拉一大堆,说实话,当我第一次看到这些文章的时候已经被这些名词给绕晕了,用了很长的时间才理清楚它们之间的关系。可能是我太蠢了,境界不够,领会不到那么多高大上的名词.
今天我用两根水管代替观察者和被观察者, 试图用通俗易懂的话把它们的关系解释清楚, 在这里我将从事件流这个角度来说明RxJava的基本工作原理。
正文
先假设有两根水管:

Single

在 Android 中使用 RxJava 最常见的场景就是网络请求,你可能使用 Retrofit 作为项目的 Http client。假设你有一个 GET HTTP 请求返回一些数据,同时使用 RxJavaAdapter 你大概会这么写:

public interface APIClient {

    @GET("my/api/path")
    Observable<MyData> getMyData();
}

上面的代码没什么问题,当调用它时:

apiClient.getMyData()
    .subscribe(new Consumer<MyData myData>() {
        @Override
        public void accept(MyData myData) throws Exception {
            // handle data fetched successfully
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception{
            // handle error event
        }
    }, new Action() {
        @Override
        public void run() throws Exception {
            // handle on complete event
        }
    });

仔细思考下,其实这个网络请求并不是一个连续事件流,你只会发起一次 Get 请求返回数据并且只收到一个事件。我们都知道这种情况下 onComplete 会紧跟着 onNext 被调用,那为什么不把它们合二为一呢?

在上面这种情况下为了更清楚的体现请求的意图,应该用Single<MyData>替换 Observable。从官方文档中对 Single 的说明可以发现为什么它是最恰当的选择:A Single is something like an Observable, but instead of emitting a series of values — anywhere from none at all to an infinite number — it always either emits one value or an error notification。所以修改后 API client 是这样的:

public interface APIClient {

    @GET("my/api/path")
    Single<MyData> getMyData();
}

同时请求的调用也可以简化:

apiClient.getMyData()
    .subscribe(new Consumer<MyData>() {
        @Override
        public void accept(MyData myData) throws Exception {
            // handle data fetched successfully and API call completed
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception{
            // handle error event
        }
    });

最值得高兴的是 Single 基本上实现了 Observable 所有的操作符 —— mapflatMapfilterzip等,如果你发现需要用到一个 Observable 的操作符而 Single 并不支持,你可以用toObservable操作符把Single<T>转换为Observable<T>

apiClient.getMyData()
    .toObservable()
    // This is an Observable<MyData> now

如果你有 Observable 表现地像 Single 一样,也可以通过singleOrError操作符转换为 Single。

Single

从SingleEmitter的源码可以看出,Single 只有 onSuccess 和 onError 事件。

/**
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing permissions and limitations under the License.
 */

package io.reactivex;

import io.reactivex.annotations.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Cancellable;

/**
 * Abstraction over an RxJava {@link SingleObserver} that allows associating
 * a resource with it.
 * <p>
 * All methods are safe to call from multiple threads.
 * <p>
 * Calling onSuccess or onError multiple times has no effect.
 *
 * @param <T> the value type to emit
 */
public interface SingleEmitter<T> {

    /**
     * Signal a success value.
     * @param t the value, not null
     */
    void onSuccess(@NonNull T t);

    /**
     * Signal an exception.
     * @param t the exception, not null
     */
    void onError(@NonNull Throwable t);

    /**
     * Sets a Disposable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param s the disposable, null is allowed
     */
    void setDisposable(@Nullable Disposable s);

    /**
     * Sets a Cancellable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param c the cancellable resource, null is allowed
     */
    void setCancellable(@Nullable Cancellable c);

    /**
     * Returns true if the downstream cancelled the sequence.
     * @return true if the downstream cancelled the sequence
     */
    boolean isDisposed();
}

其中,onSuccess()用于发射数据(在Observable/Flowable中使用onNext()来发射数据)。而且只能发射一个数据,后面即使再发射数据也不会做任何处理。

Single的SingleObserver中只有onSuccess、onError,并没有onComplete。这是 Single 跟其他四种被观察者最大的区别。

Single.create(new SingleOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {

                e.onSuccess("test");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                System.out.println(s);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                throwable.printStackTrace();
            }
        });

上面的代码,由于Observer中有两个Consumer,还可以进一步简化成

Single.create(new SingleOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {

                e.onSuccess("test");
            }
        }).subscribe(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) throws Exception {
                System.out.println(s);
            }
        });

Single 可以通过toXXX方法转换成Observable、Flowable、Completable以及Maybe。

美高梅4688官方网站 1

Completable

继续讨论 Retrofit 的例子,再看看另外一种常用场景 —— 通过 PUT 请求更新数据。我们修改了 MyData 类型对象的一些属性,把它发送到服务器更新服务器数据库。大部分服务器 API 设计都是成功后返回更新后的对象,所以你的 API client 的实现是:

public interface APIClient {

    @PUT("my/api/updatepath")
    Observable<MyData> updateMyData(@Body MyData data);
}

同样的,跟之前的例子类似,应该这样调用:

apiClient.updateMyData(myUpdatedData)
    .subscribe(new Consumer<MyData myData>() {
        @Override
        public void accept(MyData myData) throws Exception {
            // handle data fetched successfully and API call completed
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception{
            // handle error event
        }
    }, new Action() {
        @Override
        public void run() throws Exception {
            // handle completion - what we actually care about
        }
    });

你可能会说这里我们可以同样用 Single 来简化代码,是的没错。在这种情况下我们仍然需要 MyData 结果,确定?服务器返回给我们更新后的数据是良好的设计,当时实际上仅仅是返回给我们之前发送给它的对象。我们真正需要的只是更新成功了,这意味着,我只关心 onComplete 事件。

这也是引入Completable的原因,官方文档对它的描述是:Represents a computation without any value but only indication for completion or exception。使用 Completable 时我们忽略 onNext 事件,只处理 onComplete 和 onError 事件,API client 改写为:

public interface APIClient {

    @PUT("my/api/updatepath")
    Completable updateMyData(@Body MyData data);
}

调用为:

apiClient.updateMyData(myUpdatedData)
    .subscribe(new Action() {
        @Override
        public void run() throws Exception {
            // handle completion
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception{
            // handle error
        }
    });

Completable 本质上来说和 Observable 与 Single 不一样,因为它不发射数据。因此 Completable 的操作符也有所区别,最常用的是andThen。在这个操作符中你可以传任何ObservableSingleFlowableMaybe或者其他Completable,它们会在原来的 Completable 结束后执行。例如。你想执行一些其他操作(Single):

apiClient.updateMyData(myUpdatedData)
    .andThen(performOtherOperation()) // a Single<OtherResult>
    .subscribe(new Consumer<OtherResult>() {
        @Override
        public void accept(OtherResult result) throws Exception {
            // handle otherResult
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception{
            // handle error
        }
    });

跟 Single 不同的是 RxJava 不允许直接把 Observable 转换为 Completable,因为没办法知道一个 Observable 什么时候 complete。但是你可以把 Single 转换为 Completable,因为 Single 保证 onComplete 会被调用,这个操作符是toCompletable

希望通过这篇简短的对 Single 和 Completable 的介绍能让你理解这两个概念从而写出更简洁的代码。

Completable

Completable在创建后,不会发射任何数据。从CompletableEmitter的源码可以看到

/**
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing permissions and limitations under the License.
 */

package io.reactivex;

import io.reactivex.annotations.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Cancellable;

/**
 * Abstraction over an RxJava {@link CompletableObserver} that allows associating
 * a resource with it.
 * <p>
 * All methods are safe to call from multiple threads.
 * <p>
 * Calling onComplete or onError multiple times has no effect.
 */
public interface CompletableEmitter {

    /**
     * Signal the completion.
     */
    void onComplete();

    /**
     * Signal an exception.
     * @param t the exception, not null
     */
    void onError(@NonNull Throwable t);

    /**
     * Sets a Disposable on this emitter; any previous Disposable
     * or Cancellation will be disposed/cancelled.
     * @param d the disposable, null is allowed
     */
    void setDisposable(@Nullable Disposable d);

    /**
     * Sets a Cancellable on this emitter; any previous Disposable
     * or Cancellation will be disposed/cancelled.
     * @param c the cancellable resource, null is allowed
     */
    void setCancellable(@Nullable Cancellable c);

    /**
     * Returns true if the downstream disposed the sequence.
     * @return true if the downstream disposed the sequence
     */
    boolean isDisposed();
}

Completable 只有 onComplete 和 onError 事件,同时 Completable 并没有map、flatMap等操作符,它的操作符比起 Observable/Flowable 要少得多。

我们可以通过fromXXX操作符来创建一个Completable。这是一个Completable版本的Hello World。

Completable.fromAction(new Action() {
            @Override
            public void run() throws Exception {

                System.out.println("Hello World");
            }
        }).subscribe();

Completable 经常会结合andThen操作符

Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(@NonNull CompletableEmitter emitter) throws Exception {

                try {
                    TimeUnit.SECONDS.sleep(1);
                    emitter.onComplete();
                } catch (InterruptedException e) {
                    emitter.onError(e);
                }
            }
        }).andThen(Observable.range(1, 10))
        .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                System.out.println(integer);
            }
        });

在这里emitter.onComplete()执行完之后,表明Completable已经完全执行完毕,接下来是执行andThen里的操作。

打印结果如下:

1
2
3
4
5
6
7
8
9
10

在Completable中,andThen有多个重载的方法,正好对应了五种被观察者的类型。

Completable       andThen(CompletableSource next)
<T> Maybe<T>      andThen(MaybeSource<T> next)
<T> Observable<T> andThen(ObservableSource<T> next)
<T> Flowable<T>   andThen(Publisher<T> next)
<T> Single<T>     andThen(SingleSource<T> next)

Completable 也可以通过toXXX方法转换成Observable、Flowable、Single以及Maybe。

在网络操作中,如果遇到更新的情况,也就是Restful架构中的PUT操作,一般要么返回原先的对象要么只提示更新成功。下面两个接口使用了Retrofit,分别是用于获取短信验证码和更新用户信息,其中更新用户信息如果用PUT会更符合Restful的API。

    /**
     * 获取短信验证码
     * @param param
     * @return
     */
    @POST("v1/user-auth")
    Completable getVerificationCode(@Body VerificationCodeParam param);

    /**
     * 用户信息更新接口
     * @param param
     * @return
     */
    @POST("v1/user-update")
    Completable update(@Body UpdateParam param);

在model类中大致会这样写。

/**
 * Created by Tony Shen on 2017/7/24.
 */

public class VerificationCodeModel extends HttpResponse {

    /**
     * 获取验证码
     * @param activity
     * @param param
     * @return
     */
    public Completable getVerificationCode(AppCompatActivity activity, VerificationCodeParam param) {

        return apiService
                .getVerificationCode(param)
                .compose(RxJavaUtils.<VerificationCodeModel>completableToMain())
                .compose(RxLifecycle.bind(activity).<VerificationCodeModel>toLifecycleTransformer());
    }
}

特别要注意的是getVerificationCode返回的是Completable而不是Completable<T>。

获取验证码成功则给出相应地toast提示,失败可以做出相应地处理。

VerificationCodeModel model = new VerificationCodeModel();
model.getVerificationCode(RegisterActivity.this,param)
           .subscribe(new Action() {
                      @Override
                      public void run() throws Exception {
                              showShort(RegisterActivity.this,"发送验证码成功");
                      }
            },new RxException<Throwable>(){
                      @Override
                     public void accept(@NonNull Throwable throwable) throws Exception {
                              throwable.printStackTrace();
                              ......
                     }
            });

获取手机验证码.jpeg

1.1

Maybe

Maybe 是 RxJava2.x 之后才有的新类型,可以看成是Single和Completable的结合。

Maybe创建之后,MaybeEmitter 和 SingleEmitter 一样并没有onNext()方法,同样需要通过onSuccess()方法来发射数据。

Maybe.create(new MaybeOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
                e.onSuccess("testA");
            }
        }).subscribe(new Consumer<String>() {

            @Override
            public void accept(@NonNull String s) throws Exception {

                System.out.println("s="+s);
            }
        });

打印出来的结果是

s=testA

Maybe也只能发射0或者1个数据,即使发射多个数据,后面发射的数据也不会处理。

Maybe.create(new MaybeOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
                e.onSuccess("testA");
                e.onSuccess("testB");
            }
        }).subscribe(new Consumer<String>() {

            @Override
            public void accept(@NonNull String s) throws Exception {

                System.out.println("s="+s);
            }
        });

打印出来的结果仍然是

s=testA

跟第一次执行的结果是一致的。

如果MaybeEmitter先调用了onComplete(),即使后面再调用了onSuccess()也不会发射任何数据。

Maybe.create(new MaybeOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
                e.onComplete();
                e.onSuccess("testA");
            }
        }).subscribe(new Consumer<String>() {

            @Override
            public void accept(@NonNull String s) throws Exception {

                System.out.println("s="+s);
            }
        });

这次就没有打印任何数据了。

我们对上面的代码再做一下修改,在subscribe()中也加入onComplete(),看看打印出来的结果会是这样的?因为SingleObserver中是没有onComplete()方法。

Maybe.create(new MaybeOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
                e.onComplete();
                e.onSuccess("testA");
            }
        }).subscribe(new Consumer<String>() {

            @Override
            public void accept(@NonNull String s) throws Exception {

                System.out.println("s=" + s);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {

            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("Maybe onComplete");
            }
        });

这次打印的结果是

Maybe onComplete

通过查看Maybe相关的源码

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer<? super T> onSuccess, Consumer<? super Throwable> onError,
            Action onComplete) {
        ObjectHelper.requireNonNull(onSuccess, "onSuccess is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        return subscribeWith(new MaybeCallbackObserver<T>(onSuccess, onError, onComplete));
    }

我们可以得到,Maybe在没有数据发射时候subscribe会调用MaybeObserver的onComplete()。如果Maybe有数据发射或者调用了onError(),是不会再执行MaybeObserver的onComplete()。

我们也可以将 Maybe 转换成Observable、Flowable、Single,只需相应地调用toObservable()、toFlowable()、toSingle()。

接下来我们再来看看 Maybe 跟 Retrofit 是怎样结合使用的?
下面的网络请求,最初返回的类型是Flowable,但是这个网络请求并不是一个连续事件流,我们只会发起一次 Post 请求返回数据并且只收到一个事件。因此,可以考虑将 onComplete() 可以跟 onNext() 合并。在这里,尝试我们将Flowable改成Maybe。

    @POST("v1/contents")
    Maybe<ContentModel> loadContent(@Body ContentParam param);

在model类中,我们大致会这样写。

public class ContentModel extends HttpResponse {

    public List<ContentItem> data;

    /**
     * 获取内容
     * @param fragment
     * @param param
     * @param cacheKey
     * @return
     */
    public Maybe<ContentModel> getContent(Fragment fragment,ContentParam param,String cacheKey) {

        return apiService.loadContent(param)
                .compose(RxLifecycle.bind(fragment).<ContentModel>toLifecycleTransformer())
                .compose(RxJavaUtils.<ContentModel>maybeToMain())
                .compose(RxUtils.<ContentModel>toCacheTransformer(cacheKey,App.getInstance().cache));
    }

    ......
}

其中,maybeToMain()方法是用Kotlin编写的工具方法,这些工具方法由Kotlin来编写会显得比较简单和清晰,特别是lambda表达式更加直观。

    @JvmStatic
    fun <T> maybeToMain(): MaybeTransformer<T, T> {

        return MaybeTransformer{
            upstream ->
            upstream.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
        }
    }

最后是真正地使用model类,如果网络请求成功则将数据展示到recyclerview上,如果失败也会做出相应地处理。

        model.getContent(this,param,cacheKey)
                .subscribe(new Consumer<ContentModel>() {
                    @Override
                    public void accept(@io.reactivex.annotations.NonNull ContentModel model) throws Exception {

                        adapter = new NewsAdapter(mContext, model);
                        recyclerview.setAdapter(adapter);
                        spinKitView.setVisibility(View.GONE);
                    }
                }, new RxException<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        throwable.printStackTrace();
                        spinKitView.setVisibility(View.GONE);

                        ......
                    }
                });

获取内容的request.jpeg

获取内容的response.jpeg

1.1

总结

RxJava 有五种不同类型的被观察者,合理地使用它们能够写出更简洁优雅的代码。这些被观察者在一定程度上也能够作一些相互转换。值得注意的是,只有Flowable是支持Backpressure的,其余四种都不支持。

上面一根水管为事件产生的水管,叫它上游吧,下面一根水管为事件接收的水管叫它下游吧。
两根水管通过一定的方式连接起来,使得上游每产生一个事件,下游就能收到该事件。注意这里和官网的事件图是反过来的, 这里的事件发送的顺序是先1,后2,后3这样的顺序, 事件接收的顺序也是先1,后2,后3的顺序, 我觉得这样更符合我们普通人的思维, 简单明了.
这里的上游下游就分别对应着RxJava中的ObservableObserver,它们之间的连接就对应着subscribe(),因此这个关系用RxJava来表示就是:

//创建一个上游 Observable:
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
});
//创建一个下游 Observer
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}

@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}

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

@Override
public void onComplete() {
Log.d(TAG, "complete");
}
};
//建立连接
observable.subscribe(observer);

这个运行的结果就是:

12-02 03:37:17.818 4166-4166/zlc.season.rxjava2demo D/TAG: subscribe
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 1
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 2
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 3
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: complete

注意: 只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了 subscribe() 方法之后才开始发送事件.

把这段代码连起来写就成了RxJava引以为傲的链式操作:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}

@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}

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

@Override
public void onComplete() {
Log.d(TAG, "complete");
}
});

接下来解释一下其中两个陌生的玩意: ObservableEmitterDisposable.
ObservableEmitterEmitter 是发射器的意思,那就很好猜了,这个就是用来发出事件的,它可以发出三种类型的事件,通过调用emitter的 onNext(T value)onComplete()onError(Throwable error)就可以分别发出next事件、complete事件和error事件。但是,请注意,并不意味着你可以随意乱七八糟发射事件,需要满足一定的规则:
上游可以发送无限个onNext, 下游也可以接收无限个onNext.
当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
上游可以不发送onComplete或onError.
最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然注: 关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则, 并不一定会导致程序崩溃. 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.

以上几个规则用示意图表示如下:| Action | 示意图 || —————– | ———– || 只发送onNext事件 |

美高梅4688官方网站 2

next

next

|| 发送onComplete事件 |

美高梅4688官方网站 3

complete

complete

|| 发送onError事件 |

美高梅4688官方网站 4

error

error

|
介绍了ObservableEmitter, 接下来介绍Disposable, 这个单词的字面意思是一次性用品,用完即可丢弃的. 那么在RxJava中怎么去理解它呢, 对应于上面的水管的例子, 我们可以把它理解成两根管道之间的一个机关, 当调用它的 dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件.
注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.

来看个例子, 我们让上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 切断水管, 看看运行结果:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit complete");
emitter.onComplete();
Log.d(TAG, "emit 4");
emitter.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDisposable;
private int i;

@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
mDisposable = d;
}

@Override
public void onNext(Integer value) {
Log.d(TAG, "onNext: " + value);
i++;
if (i == 2) {
Log.d(TAG, "dispose");
mDisposable.dispose();
Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
}
}

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

@Override
public void onComplete() {
Log.d(TAG, "complete");
}
});

运行结果为:

12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: subscribe
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: dispose
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: isDisposed : true
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 3
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit complete
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 4

从运行结果我们看到, 在收到onNext 2这个事件后, 切断了水管, 但是上游仍然发送了3, complete, 4这几个事件, 而且上游并没有因为发送了onComplete而停止. 同时可以看到下游的 onSubscribe()方法是最先调用的.Disposable的用处不止这些, 后面讲解到了线程的调度之后, 我们会发现它的重要性. 随着后续深入的讲解, 我们会在更多的地方发现它的身影.另外, subscribe()有多个重载的方法:

public final Disposable subscribe() {}
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}

最后一个带有Observer参数的我们已经使用过了,这里对其他几个方法进行说明.
不带任何参数的 subscribe()表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见, 因此我们如果只需要onNext事件可以这么写:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit complete");
emitter.onComplete();
Log.d(TAG, "emit 4");
emitter.onNext(4);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "onNext: " + integer);
}
});

其他几个方法同理, 这里就不一一解释了.

第二式 飞龙在天
概述
上一节教程讲解了最基本的RxJava2的使用, 在本节中, 我们将学习RxJava强大的线程控制.
正文
还是以之前的例子, 两根水管:

美高梅4688官方网站 5

2.1

2.1

正常情况下, 上游和下游是工作在同一个线程中的, 也就是说上游在哪个线程发事件, 下游就在哪个线程接收事件.
在RxJava中, 当我们在主线程中去创建一个上游Observable来发送事件, 则这个上游默认就在主线程发送事件.
当我们在主线程去创建一个下游Observer来接收事件, 则这个下游默认就在主线程中接收事件, 来看段代码:

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
Log.d(TAG, "emit 1");
emitter.onNext(1);
}
});

Consumer<Integer> consumer = new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
Log.d(TAG, "onNext: " + integer);
}
};

observable.subscribe(consumer);
}

在主线程中分别创建上游和下游, 然后将他们连接在一起, 同时分别打印出它们所在的线程, 运行结果为:
D/TAG: Observable thread is : main
D/TAG: emit 1
D/TAG: Observer thread is :main
D/TAG: onNext: 1

这就验证了刚才所说, 上下游默认是在同一个线程工作.
这样肯定是满足不了我们的需求的, 我们更多想要的是这么一种情况, 在子线程中做耗时的操作, 然后回到主线程中来操作UI, 用图片来描述就是下面这个图片:

美高梅4688官方网站 6

2.2

2.2

在这个图中, 我们用黄色水管表示子线程, 深蓝色水管表示主线程.
要达到这个目的, 我们需要先改变上游发送事件的线程, 让它去子线程中发送事件, 然后再改变下游的线程, 让它去主线程接收事件. 通过RxJava内置的线程调度器可以很轻松的做到这一点. 接下来看一段代码:

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
Log.d(TAG, "emit 1");
emitter.onNext(1);
}
});

Consumer<Integer> consumer = new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
Log.d(TAG, "onNext: " + integer);
}
};

observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
}

还是刚才的例子, 只不过我们太添加了一点东西, 先来看看运行结果:

D/TAG: Observable thread is : RxNewThreadScheduler-2
D/TAG: emit 1
D/TAG: Observer thread is :main
D/TAG: onNext: 1

可以看到, 上游发送事件的线程的确改变了, 是在一个叫 RxNewThreadScheduler-2的线程中发送的事件, 而下游仍然在主线程中接收事件, 这说明我们的目的达成了, 接下来看看是如何做到的.和上一段代码相比,这段代码只不过是增加了两行代码:

.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())

简单的来说, subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.

在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程

这些内置的Scheduler已经足够满足我们开发的需求, 因此我们应该使用内置的这些选项, 在RxJava内部使用的是线程池来维护这些线程, 所有效率也比较高.
实践
对于我们Android开发人员来说, 经常会将一些耗时的操作放在后台, 比如网络请求或者读写文件,操作数据库等等,等到操作完成之后回到主线程去更新UI, 有了上面的这些基础, 那么现在我们就可以轻松的去做到这样一些操作.
下面来举几个常用的场景.
网络请求
Android中有名的网络请求库就那么几个, Retrofit能够从中脱颖而出很大原因就是因为它支持RxJava的方式来调用, 下面简单讲解一下它的基本用法.
要使用 Retrofit,先添加Gradle配置,最新的版本可在GitHub找到。Retrofit、Gson converter、RxJava2 Adapter、okhttp、logging-interceptor

//retrofit
compile 'com.squareup.retrofit2:retrofit:2.2.0'
//Gson converter
compile 'com.squareup.retrofit2:converter-gson:2.2.0'
//RxJava2 Adapter
compile 'com.squareup.retrofit2:adapter-rxjava2:2.2.0'
//okhttp
compile 'com.squareup.okhttp3:okhttp:3.6.0'
compile 'com.squareup.okhttp3:logging-interceptor:3.6.0'

随后定义Api接口:

public interface Api {
@GET
Observable<LoginResponse> login(@Body LoginRequest request);

@GET
Observable<RegisterResponse> register(@Body RegisterRequest request);
}

接着创建一个Retrofit客户端:

private static Retrofit create() {
OkHttpClient.Builder builder = new OkHttpClient().newBuilder();
builder.readTimeout(10, TimeUnit.SECONDS);
builder.connectTimeout(9, TimeUnit.SECONDS);

if (BuildConfig.DEBUG) {
HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor();
interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
builder.addInterceptor(interceptor);
}

return new Retrofit.Builder().baseUrl(ENDPOINT)
.client(builder.build())
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
.build();
}

发起请求就很简单了:

Api api = retrofit.create(Api.class);
api.login(request)
.subscribeOn(Schedulers.io()) //在IO线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) //回到主线程去处理请求结果
.subscribe(new Observer<LoginResponse>() {
@Override
public void onSubscribe(Disposable d) {}

@Override
public void onNext(LoginResponse value) {}

@Override
public void onError(Throwable e) {
Toast.makeText(mContext, "登录失败", Toast.LENGTH_SHORT).show();
}

@Override
public void onComplete() {
Toast.makeText(mContext, "登录成功", Toast.LENGTH_SHORT).show();
}
});

看似很完美, 但我们忽略了一点, 如果在请求的过程中Activity已经退出了, 这个时候如果回到主线程去更新UI, 那么APP肯定就崩溃了, 怎么办呢, 上一节我们说到了 Disposable , 说它是个开关, 调用它的dispose()方法时就会切断水管, 使得下游收不到事件, 既然收不到事件, 那么也就不会再去更新UI了. 因此我们可以在Activity中将这个Disposable 保存起来, 当Activity退出时, 切断它即可.
那如果有多个Disposable 该怎么办呢, RxJava中已经内置了一个容器 CompositeDisposable, 每当我们得到一个Disposable时就调用 CompositeDisposable.add()将它添加到容器中, 在退出的时候, 调用CompositeDisposable.clear() 即可切断所有的水管.

读写数据库
上面说了网络请求的例子, 接下来再看看读写数据库, 读写数据库也算一个耗时的操作, 因此我们也最好放在IO线程里去进行, 这个例子就比较简单, 直接上代码:

public Observable<List<Record>> readAllRecords() {
return Observable.create(new ObservableOnSubscribe<List<Record>>() {
@Override
public void subscribe(ObservableEmitter<List<Record>> emitter) throws Exception {
Cursor cursor = null;
try {
cursor = getReadableDatabase().rawQuery("select * from " + TABLE_NAME, new String[]{});
List<Record> result = new ArrayList<>();
while (cursor.moveToNext()) {
result.add(Db.Record.read(cursor));
}
emitter.onNext(result);
emitter.onComplete();
} finally {
if (cursor != null) {
cursor.close();
}
}
}
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

好了节就到这里吧, 后面将会教大家如何使用RxJava中强大的操作符. 通过使用这些操作符可以很轻松的做到各种吊炸天的效果.
第三式 见龙在田
概述
上一节讲解了线程调度, 并且举了两个实际中的例子, 其中有一个登录的例子, 不知大家有没有想过这么一个问题, 如果是一个新用户, 必须先注册, 等注册成功之后再自动登录该怎么做呢.很明显, 这是一个嵌套的网络请求, 首先需要去请求注册, 待注册成功回调了再去请求登录的接口.我们当然可以想当然的写成这样:

private void login() {
api.login(new LoginRequest())
.subscribeOn(Schedulers.io()) //在IO线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) //回到主线程去处理请求结果
.subscribe(new Consumer<LoginResponse>() {
@Override
public void accept(LoginResponse loginResponse) throws Exception {
Toast.makeText(MainActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
}
});
}

private void register() {
api.register(new RegisterRequest())
.subscribeOn(Schedulers.io()) //在IO线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) //回到主线程去处理请求结果
.subscribe(new Consumer<RegisterResponse>() {
@Override
public void accept(RegisterResponse registerResponse) throws Exception {
Toast.makeText(MainActivity.this, "注册成功", Toast.LENGTH_SHORT).show();
login(); //注册成功, 调用登录的方法
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, "注册失败", Toast.LENGTH_SHORT).show();
}
});
}

这样的代码能够工作, 但不够优雅, 通过本节的学习, 可以让我们用一种更优雅的方式来解决这个问题.
正文
先来看看最简单的变换操作符map吧.
Map
map是RxJava中最简单的一个变换操作符了, 它的作用就是对上游发送的每一个事件应用一个函数, 使得每一个事件都按照指定的函数去变化. 用事件图表示如下:

美高梅4688官方网站 7

map

map

图中map中的函数作用是将圆形事件转换为矩形事件, 从而导致下游接收到的事件就变为了矩形.用代码来表示这个例子就是:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "This is result " + integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
});

在上游我们发送的是数字类型, 而在下游我们接收的是String类型, 中间起转换作用的就是map操作符, 运行结果为:
D/TAG: This is result 1
D/TAG: This is result 2
D/TAG: This is result 3

通过Map, 可以将上游发来的事件转换为任意的类型, 可以是一个Object, 也可以是一个集合, 如此强大的操作符你难道不想试试?接下来我们来看另外一个广为人知的操作符flatMap.
FlatMap
flatMap是一个非常强大的操作符, 先用一个比较难懂的概念说明一下:FlatMap将一个发送事件的上游Observable变换为多个发送事件的Observables,然后将它们发射的事件合并后放进一个单独的Observable里.这句话比较难以理解, 我们先通俗易懂的图片来详细的讲解一下, 首先先来看看整体的一个图片:

美高梅4688官方网站 8

flatMap

flatMap

先看看上游, 上游发送了三个事件, 分别是1,2,3, 注意它们的颜色.中间flatMap的作用是将圆形的事件转换为一个发送矩形事件和一个三角形事件的新的上游Observable.
上游每发送一个事件, flatMap都将创建一个新的水管, 然后发送转换之后的新的事件, 下游接收到的就是这些新的水管发送的数据. 这里需要注意的是, flatMap并不保证事件的顺序, 也就是图中所看到的, 并不是事件1就在事件2的前面. 如果需要保证顺序则需要使用concatMap.说了原理, 我们还是来看看实际中的代码如何写吧:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
});

如代码所示, 我们在flatMap中将上游发来的每个事件转换为一个新的发送三个String事件的水管, 为了看到flatMap结果是无序的,所以加了10毫秒的延时, 来看看运行结果吧:

D/TAG: I am value 1
D/TAG: I am value 1
D/TAG: I am value 1
D/TAG: I am value 3
D/TAG: I am value 3
D/TAG: I am value 3
D/TAG: I am value 2
D/TAG: I am value 2
D/TAG: I am value 2

concatMap
这里也简单说一下concatMap吧, 它和flatMap的作用几乎一模一样, 只是它的结果是严格按照上游发送的顺序来发送的, 来看个代码吧:

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
return Observable.fromIterable(list).delay(10,TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
});

只是将之前的flatMap改为了concatMap, 其余原封不动, 运行结果如下:
D/TAG: I am value 1
D/TAG: I am value 1
D/TAG: I am value 1
D/TAG: I am value 2
D/TAG: I am value 2
D/TAG: I am value 2
D/TAG: I am value 3
D/TAG: I am value 3
D/TAG: I am value 3

可以看到, 结果仍然是有序的.

好了关于RxJava的操作符最基本的使用就讲解到这里了, RxJava中内置了许许多多的操作符, 这里通过讲解map和flatMap只是起到一个抛砖引玉的作用, 关于其他的操作符只要大家按照本文的思路去理解, 再仔细阅读文档, 应该是没有问题的了.
实践
如何优雅的解决嵌套请求, 只需要用flatMap转换一下就行了.先回顾一下上一节的请求接口:

public interface Api {
@GET
Observable<LoginResponse> login(@Body LoginRequest request);

@GET
Observable<RegisterResponse> register(@Body RegisterRequest request);
}

可以看到登录和注册返回的都是一个上游Observable, 而我们的flatMap操作符的作用就是把一个Observable转换为另一个Observable, 因此结果就很显而易见了:

api.register(new RegisterRequest()) //发起注册请求
.subscribeOn(Schedulers.io()) //在IO线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) //回到主线程去处理请求注册结果
.doOnNext(new Consumer<RegisterResponse>() {
@Override
public void accept(RegisterResponse registerResponse) throws Exception {
//先根据注册的响应结果去做一些操作
}
})
.observeOn(Schedulers.io()) //回到IO线程去发起登录请求
.flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() {
@Override
public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
return api.login(new LoginRequest());
}
})
.observeOn(AndroidSchedulers.mainThread()) //回到主线程去处理请求登录的结果
.subscribe(new Consumer<LoginResponse>() {
@Override
public void accept(LoginResponse loginResponse) throws Exception {
Toast.makeText(MainActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
}
});

从这个例子也可以看到我们切换线程是多么简单.下一节我们将会学到 zip 操作符.
第四式 鸿渐于陆
概述
下面学习zip这个操作符, 这个操作符也是比较牛逼的东西了, 涉及到的东西也比较多, 主要是一些细节上的东西太多, 通过学习这个操作符,可以为我们下一节的Backpressure做个铺垫.
正文
Zip通过一个函数将多个Observable发送的事件结合到一起,然后发送这些组合到一起的事件. 它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。我们再用通俗易懂的图片来解释一下:

美高梅4688官方网站 9

zip

zip

从这个图中可以看见, 这次上游和以往不同的是, 我们有两根水管了.其中一根水管负责发送圆形事件 , 另外一根水管负责发送三角形事件 , 通过Zip操作符, 使得圆形事件和三角形事件 合并为了一个矩形事件.
组合的过程是分别从 两根水管里各取出一个事件 来进行组合, 并且一个事件只能被使用一次, 组合的顺序是严格按照事件发送的顺利 来进行的, 也就是说不会出现圆形1 事件和三角形B 事件进行合并, 也不可能出现圆形2 和三角形A 进行合并的情况.
最终下游收到的事件数量 是和上游中发送事件最少的那一根水管的事件数量 相同. 这个也很好理解, 因为是从每一根水管 里取一个事件来进行合并, 最少的 那个肯定就最先取完 , 这个时候其他的水管尽管还有事件 , 但是已经没有足够的事件来组合了, 因此下游就不会收到剩余的事件了.

分析了大概的原理, 我们还是劳逸结合, 先来看看实际中的代码怎么写吧:

Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit 4");
emitter.onNext(4);
Log.d(TAG, "emit complete1");
emitter.onComplete();
}
});

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "emit A");
emitter.onNext("A");
Log.d(TAG, "emit B");
emitter.onNext("B");
Log.d(TAG, "emit C");
emitter.onNext("C");
Log.d(TAG, "emit complete2");
emitter.onComplete();
}
});

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe");
}

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

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

@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});

我们分别创建了两个上游水管, 一个发送1,2,3,4,Complete, 另一个发送A,B,C,Complete, 接着用Zip把发出的事件组合, 来看看运行结果吧:

D/TAG: onSubscribe
D/TAG: emit 1
D/TAG: emit 2
D/TAG: emit 3
D/TAG: emit 4
D/TAG: emit complete1
D/TAG: emit A
D/TAG: onNext: 1A
D/TAG: emit B
D/TAG: onNext: 2B
D/TAG: emit C
D/TAG: onNext: 3C
D/TAG: emit complete2
D/TAG: onComplete

结果似乎是对的… 但是总感觉什么地方不对劲…哪儿不对劲呢, 为什么感觉是水管一发送完了之后, 水管二才开始发送啊? 到底是不是呢, 我们来验证一下:

Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Thread.sleep(1000);

Log.d(TAG, "emit 2");
emitter.onNext(2);
Thread.sleep(1000);

Log.d(TAG, "emit 3");
emitter.onNext(3);
Thread.sleep(1000);

Log.d(TAG, "emit 4");
emitter.onNext(4);
Thread.sleep(1000);

Log.d(TAG, "emit complete1");
emitter.onComplete();
}
});

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "emit A");
emitter.onNext("A");
Thread.sleep(1000);

Log.d(TAG, "emit B");
emitter.onNext("B");
Thread.sleep(1000);

Log.d(TAG, "emit C");
emitter.onNext("C");
Thread.sleep(1000);

Log.d(TAG, "emit complete2");
emitter.onComplete();
}
});

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe");
美高梅4688官方网站,}

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

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

@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});

这次我们在每发送一个事件之后加入了一秒钟的延时, 来看看运行结果吧, 注意这是个GIF图:

美高梅4688官方网站 10

zip

zip

好像真的是先发送的水管一再发送的水管二呢, 为什么会有这种情况呢? 因为我们两根水管都是运行在同一个线程里, 同一个线程里执行代码肯定有先后顺序呀.
因此我们来稍微改一下, 不让他们在同一个线程, 不知道怎么切换线程的, 请掉头看前面几节.

Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Thread.sleep(1000);

Log.d(TAG, "emit 2");
emitter.onNext(2);
Thread.sleep(1000);

Log.d(TAG, "emit 3");
emitter.onNext(3);
Thread.sleep(1000);

Log.d(TAG, "emit 4");
emitter.onNext(4);
Thread.sleep(1000);

Log.d(TAG, "emit complete1");
emitter.onComplete();
}
}).subscribeOn(Schedulers.io());

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "emit A");
emitter.onNext("A");
Thread.sleep(1000);

Log.d(TAG, "emit B");
emitter.onNext("B");
Thread.sleep(1000);

Log.d(TAG, "emit C");
emitter.onNext("C");
Thread.sleep(1000);

Log.d(TAG, "emit complete2");
emitter.onComplete();
}
}).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe");
}

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

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

@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});

好了, 这次我们让水管都在IO线程里发送事件, 再来看看运行结果:

D/TAG: onSubscribe
D/TAG: emit A
D/TAG: emit 1
D/TAG: onNext: 1A
D/TAG: emit B
D/TAG: emit 2
D/TAG: onNext: 2B
D/TAG: emit C
D/TAG: emit 3
D/TAG: onNext: 3C
D/TAG: emit complete2
D/TAG: onComplete

GIF图:

美高梅4688官方网站 11

zip

zip

这下就对了嘛, 两根水管同时开始发送, 每发送一个, Zip就组合一个, 再将组合结果发送给下游.不对呀! 可能细心点的朋友又看出端倪了, 第一根水管明明发送了四个数据+一个Complete, 之前明明还有的, 为啥到这里没了呢?这是因为我们之前说了, zip发送的事件数量跟上游中发送事件最少的那一根水管的事件数量是有关的, 在这个例子里我们第二根水管只发送了三个事件然后就发送了Complete, 这个时候尽管第一根水管还有事件4事件Complete 没有发送, 但是它们发不发送还有什么意义呢? 所以本着节约是美德的思想, 就干脆打断它的狗腿, 不让它发了.至于前面的例子为什么会发送, 刚才不是已经说了是!在!同!一!个!线!程!里!吗!!!!再问老子打死你!有好事的程序员可能又要问了, 那我不发送Complete呢? 答案是显然的, 上游会继续发送事件, 但是下游仍然收不到那些多余的事件. 不信你可以试试.
实践
学习了Zip的基本用法, 那么它在Android有什么用呢, 其实很多场景都可以用到Zip. 举个例子.比如一个界面需要展示用户的一些信息, 而这些信息分别要从两个服务器接口中获取, 而只有当两个都获取到了之后才能进行展示, 这个时候就可以用Zip了:首先分别定义这两个请求接口:

public interface Api {
@GET
Observable<UserBaseInfoResponse> getUserBaseInfo(@Body UserBaseInfoRequest request);

@GET
Observable<UserExtraInfoResponse> getUserExtraInfo(@Body UserExtraInfoRequest request);

}

接着用Zip来打包请求:

Observable<UserBaseInfoResponse> observable1 =
api.getUserBaseInfo(new UserBaseInfoRequest()).subscribeOn(Schedulers.io());

Observable<UserExtraInfoResponse> observable2 =
api.getUserExtraInfo(new UserExtraInfoRequest()).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2,
new BiFunction<UserBaseInfoResponse, UserExtraInfoResponse, UserInfo>() {
@Override
public UserInfo apply(UserBaseInfoResponse baseInfo,
UserExtraInfoResponse extraInfo) throws Exception {
return new UserInfo(baseInfo, extraInfo);
}
}).observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<UserInfo>() {
@Override
public void accept(UserInfo userInfo) throws Exception {
//do something;
}
});

本次的教程就到这里吧.下一节我们将会学到 Flowable 以及理解Backpressure背压的概念.
第五式 潜龙勿用
概述
这一节中我们将来学习Backpressure. 我看好多吃瓜群众早已坐不住了, 别急, 我们先来回顾一下上一节讲的Zip.
正文
上一节中我们说到Zip可以将多个上游发送的事件组合起来发送给下游, 那大家有没有想过一个问题, 如果其中一个水管A发送事件特别快, 而另一个水管B发送事件特别慢, 那就可能出现这种情况, 发得快的水管A 已经发送了1000个事件了, 而发的慢的水管B 才发一个出来, 组合了一个之后水管A 还剩999个事件, 这些事件需要继续等待水管B 发送事件出来组合, 那么这么多的事件是放在哪里的呢? 总有一个地方保存吧? 没错, Zip给我们的每一根水管都弄了一个水缸 , 用来保存这些事件, 用通俗易懂的图片来表示就是:

美高梅4688官方网站 12

zip

zip

如图中所示, 其中蓝色的框框就是zip给我们的水缸! 它将每根水管发出的事件保存起来, 等两个水缸都有事件了之后就分别从水缸中取出一个事件来组合, 当其中一个水缸是空的时候就处于等待的状态.题外话: 大家来分析一下这个水缸有什么特点呢? 它是按顺序保存的, 先进来的事件先取出来, 这个特点是不是很熟悉呀? 没错, 这就是我们熟知的队列, 这个水缸在Zip内部的实现就是用的队列, 感兴趣的可以翻看源码查看.好了回到正题上来, 这个水缸有大小限制吗? 要是一直往里存会怎样? 我们来看个例子:
Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) { //无限循环发事件
emitter.onNext(i);
}
}
}).subscribeOn(Schedulers.io());

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("A");
}
}).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.w(TAG, throwable);
}
});

在这个例子中, 我们分别创建了两根水管, 第一根水管用机器指令的执行速度来无限循环发送事件, 第二根水管随便发送点什么, 由于我们没有发送Complete事件, 因此第一根水管会一直发事件到它对应的水缸里去, 我们来看看运行结果是什么样.运行结果GIF图:

美高梅4688官方网站 13

zip

zip

内存占用以斜率为1的直线迅速上涨, 几秒钟就300多M , 最终报出了OOM:

zlc.season.rxjava2demo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with
4194304 free bytes and 8MB until OOM;
zlc.season.rxjava2demo W/art: "main" prio=5 tid=1 Runnable
zlc.season.rxjava2demo W/art: | group="main" sCount=0 dsCount=0 obj=0x75188710 self=0x7fc0efe7ba00
zlc.season.rxjava2demo W/art: | sysTid=32686 nice=0 cgrp=default sched=0/0 handle=0x7fc0f37dc200
zlc.season.rxjava2demo W/art: | state=R schedstat=( 0 0 0 ) utm=948 stm=120 core=1 HZ=100
zlc.season.rxjava2demo W/art: | stack=0x7fff971e8000-0x7fff971ea000 stackSize=8MB
zlc.season.rxjava2demo W/art: | held mutexes= "mutator lock"(shared held)
zlc.season.rxjava2demo W/art: at java.lang.Integer.valueOf(Integer.java:742)

出现这种情况肯定是我们不想看见的, 这里就可以引出我们的Backpressure了, 所谓的Backpressure其实就是为了控制流量, 水缸存储的能力毕竟有限, 因此我们还得从源头去解决问题, 既然你发那么快, 数据量那么大, 那我就想办法不让你发那么快呗.那么这个源头到底在哪里, 究竟什么时候会出现这种情况, 这里只是说的Zip这一个例子, 其他的地方会出现吗? 带着这个问题我们来探究一下.我们让事情变得简单一点, 从一个单一的Observable说起.来看段代码:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) { //无限循环发事件
emitter.onNext(i);
}
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
Log.d(TAG, "" + integer);
}
});

这段代码很简单, 上游同样无限循环的发送事件, 在下游每次接收事件前延时2秒. 上下游工作在同一个线程里, 来看下运行结果:

美高梅4688官方网站 14

5.3

5.3

本文由美高梅4688官方网站发布于美高梅4688官方网站,转载请注明出处:就可以应用上百个操作符来实现自己的表达式,首先会想到的是使用Observable

您可能还会对下面的文章感兴趣: