RxJava 方法备忘录

Posted 2020-07-26 一叶飘舟

RxJava 方法

过滤相关

RxJava提供了filter()方法来过滤我们观测序列中不想要的值

当我们不需要整个序列时，而是只想取开头或结尾的几个元素，我们可以用take()或takeLast()

一个可观测序列会在出错时重复发射或者被设计成重复发射。distinct()和distinctUntilChanged()函数可以方便的让我们处理这种重复问题。(它会记录以及发射的值来过滤，所以请注意内存)

first()方法和last()方法很容易弄明白。它们从Observable中只发射第一个元素或者最后一个元素。这两个都可以传Func1作为参数，：一个可以确定我们感兴趣的第一个或者最后一个的谓词：

skip()和skipLast()函数与take()和takeLast()相对应。它们用整数N作参数，从本质上来说，它们不让Observable发射前N个或者后N个值。如果我们知道一个序列以没有太多用的“可控”元素开头或结尾时我们可以使用它。

elementAt()函数仅从一个序列中发射第n个元素,如果怕不存在可以使用elementAtOrDefault()

一个温度传感器，它每秒都会发射当前室内的温度。说实话，我们并不认为温度会变化这么快，我们可以使用一个小的发射间隔。在Observable后面加一个sample()，我们将创建一个新的可观测序列，它将在一个指定的时间间隔里由Observable发射最近一次的数值：

假设我们工作的是一个时效性的环境，我们温度传感器每秒都在发射一个温度值。我们想让它每隔两秒至少发射一个，我们可以使用timeout()函数来监听源可观测序列,就是在我们设定的时间间隔内如果没有得到一个值则发射一个错误。我们可以认为timeout()为一个Observable的限时的副本。如果在指定的时间间隔内Observable不发射值的话，它监听的原始的Observable时就会触发onError()函数。

debounce()函数过滤掉由Observable发射的速率过快的数据；如果在一个指定的时间间隔过去了仍旧没有发射一个，那么它将发射最后的那个。

就像sample()和timeout()函数一样，debounce()使用TimeUnit对象指定时间间隔。

变换相关

RxJava提供了几个mapping函数：map(),flatMap(),concatMap(),flatMapIterable()以及switchMap().所有这些函数都作用于一个可观测序列，然后变换它发射的值，最后用一种新的形式返回它们

在复杂的场景中，我们有一个这样的Observable：它发射一个数据序列，这些数据本身也可以发射Observable。RxJava的flatMap()函数提供一种铺平序列的方式，然后合并这些Observables发射的数据，最后将合并后的结果作为最终的Observable。当我们在处理可能有大量的Observables时，重要是记住任何一个Observables发生错误的情况，flatMap()将会触发它自己的onError()函数并放弃整个链。

重要的一点提示是关于合并部分：它允许交叉,所以flatMap()不能够保证在最终生成的Observable中源Observables确切的发射顺序。

concatMap()函数解决了flatMap()的交叉问题，提供了一种能够把发射的值连续在一起的铺平函数，而不是合并它们，如下图所示：

witchMap()和flatMap()很像，除了一点：每当源Observable发射一个新的数据项（Observable）时，它将取消订阅并停止监视之前那个数据项产生的Observable，并开始监视当前发射的这一个

scan()函数可以看做是一个累积函数。scan()函数对原始Observable发射的每一项数据都应用一个函数，计算出函数的结果值，并将该值填充回可观测序列，等待和下一次发射的数据一起使用。

groupBy() 将源Observable变换成一个发射Observables的新的Observable。它们中的每一个新的Observable都发射一组指定的数据。

Buffer() 将源Observable变换一个新的Observable，这个新的Observable每次发射一组列表值而不是一个一个发射

window()函数和buffer()很像，但是它发射的是Observable而不是列表,数量由count指定,最后发射一个onCompleted()结束。正如buffer()一样,window()也有一个skip变体

cast()函数是map()操作符的特殊版本。它将源Observable中的每一项数据都转换为新的类型，把它变成了不同的Class。

组合Observables

merge() 我们有多个来源但是又只想有一个结果：多输入，单输出。RxJava的merge()方法将帮助你把两个甚至更多的Observables合并到他们发射的数据项里。每个Observable抛出的错误都将会打断合并。如果你需要避免这种情况，RxJava提供了mergeDelayError()，它能从一个Observable中继续发射数据即便是其中有一个抛出了错误。当所有的Observables都完成时，mergeDelayError()将会发射onError()

zip() 处理多个数据来源时会带来：多从个Observables接收数据，处理它们，然后将它们合并成一个新的可观测序列来使用。RxJava有一个特殊的方法可以完成：zip()合并两个或者多个Observables发射出的数据项，根据指定的函数Func*变换它们，并发射一个新值。

Join() 前面的zip()和merge()方法作用在发射数据的范畴内，在决定如何操作值之前有些场景我们需要考虑时间的。RxJava的join()函数基于时间窗口将两个Observables发射的数据结合在一起

combineLatest() 有点像zip()函数的特殊形式。zip()作用于最近未打包的两个Observables。相反，combineLatest()作用于最近发射的数据项：如果Observable1发射了A并且Observable2发射了B和C，combineLatest()将会分组处理AB和AC.

在将来还有一些zip()满足不了的场景。如复杂的架构，或者是仅仅为了个人爱好，你可以使用And/Then/When解决方

switch 有这样一个复杂的场景就是在一个subscribe-unsubscribe的序列里我们能够从一个Observable自动取消订阅来订阅一个新的Observable。

RxJava的switch()，正如定义的，将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项。

给出一个发射多个Observables序列的源Observable，switch()订阅到源Observable然后开始发射由第一个发射的Observable发射的一样的数据。当源Observable发射一个新的Observable时，switch()立即取消订阅前一个发射数据的Observable（因此打断了从它那里发射的数据流）然后订阅一个新的Observable，并开始发射它的数据。

startWith()是concat()的对应部分。正如concat()向发射数据的Observable追加数据那样，在Observable开始发射他们的数据之前， startWith()通过传递一个参数来先发射一个数据序列

调用线程

StrictMode帮助我们侦测敏感的活动，如我们无意的在主线程执行磁盘访问或者网络调用。正如你所知道的，在主线程执行繁重的或者长时的任务是不可取的。因为android应用的主线程时UI线程，它被用来处理和UI相关的操作：这也是获得更平滑的动画体验和响应式App的唯一方法。

为了在我们的App中激活StrictMode，我们只需要在MainActivity中添加几行代码，即onCreate()方法

@Override
public void onCreate() { 
    super.onCreate();
    if (BuildConfig.DEBUG) {
        StrictMode.setThreadPolicy(new StrictMode.ThreadPolicy.Builder().detectAll().penaltyLog().build()); 
        StrictMode.setVmPolicy(new StrictMode.VmPolicy.Builder().detectAll().penaltyLog().build());
    } 
}

我们并不想它总是激活着，因此我们只在debug构建时使用。这种配置将报告每一种关于主线程用法的违规做法，并且这些做法都可能与内存泄露有关：Activities、BroadcastReceivers、Sqlite等对象。+

选择了penaltyLog()，当违规做法发生时，StrictMode将会在logcat打印一条信息。

　　阻塞I/O的操作会导致App必须等待结果返回（阻塞结束）才能进行下一步操作。在UI线程上执行一个阻塞操作会将UI强行卡住，直接造成很糟糕的用户体验。

我们激活StrictMode后，我们开始收到了关于我们的App错误操作磁盘I/O的不良信息。

D/StrictMode  StrictMode policy violation; ~duration=998 ms: android.os.StrictMode$StrictModeDiskReadViolation: policy=31 violation=2
at android.os.StrictMode$AndroidBlockGuardPolicy.onReadFromDisk (StrictMode.java:1135)
at libcore.io.BlockGuardOs.open(BlockGuardOs.java:106) at libcore.io.IoBridge.open(IoBridge.java:393)
at java.io.FileOutputStream.<init>(FileOutputStream.java:88) 
at android.app.ContextImpl.openFileOutput(ContextImpl.java:918) 
at android.content.ContextWrapper.openFileOutput(ContextWrapper. java:185)
at com.packtpub.apps.rxjava_essentials.Utils.storeBitmap (Utils.java:30)

上一条信息告诉我们Utils.storeBitmap()函数执行完耗时998ms：在UI线程上近1秒的不必要的工作和App上近1秒不必要的迟钝。这是因为我们以阻塞的方式访问磁盘。我们的storeBitmap()函数包含了：

[code]FileOutputStream fOut = context.openFileOutput(filename, Context.MODE_PRIVATE);

它直接访问智能手机的固态存储然后就慢了。我们该如何提高访问速度呢？storeBitmap()函数保存了已安装App的图标。他的返回值类型为void，因此在执行下一个操作前我们毫无理由去等待直到它完成。我们可以启动它并让它执行在不同的线程。近几年来Android的线程管理发生了许多变化，导致App出现诡异的行为。我们可以使用AsyncTask，但是我们要避免掉入前几章里的onPre… onPost…doInBackGround地狱。下面我们将换用RxJava的方式。调度器万岁！

*

Schedulers.io() 这个调度器时用于I/O操作。它基于根据需要，增长或缩减来自适应的线程池。我们将使用它来修复我们之前看到的StrictMode违规做法。由于它专用于I/O操作，所以并不是RxJava的默认方法；正确的使用它是由开发者决定的。

重点需要注意的是线程池是无限制的，大量的I/O调度操作将创建许多个线程并占用内存。一如既往的是，我们需要在性能和简捷两者之间找到一个有效的平衡点。

Schedulers.computation() 这个是计算工作默认的调度器，它与I/O操作无关。它也是许多RxJava方法的默认调度器：buffer(),debounce(),delay(),interval(),sample(),skip()。

Schedulers.immediate()

这个调度器允许你立即在当前线程执行你指定的工作。它是timeout(),timeInterval(),以及timestamp()方法默认的调度器。

Schedulers.newThread()

这个调度器正如它所看起来的那样：它为指定任务启动一个新的线程。

Schedulers.trampoline()

当我们想在当前线程执行一个任务时，并不是立即，我们可以用.trampoline()将它入队。这个调度器将会处理它的队列并且按序运行队列中每一个任务。它是repeat()和retry()方法默认的调度器。

非阻塞I/O操作

现在我们知道如何在一个指定I/O调度器上来调度一个任务，我们可以修改storeBitmap()函数并再次检查StrictMode的不合规做法。为了这个例子，我们可以在新的blockingStoreBitmap()函数中重排代码。

    FileOutputStream fOut = null; 
    try {
        fOut = context.openFileOutput(filename, Context.MODE_PRIVATE);
        bitmap.compress(Bitmap.CompressFormat.PNG, 100, fOut); 
        fOut.flush();
        fOut.close();
    } catch (Exception e) {
        throw new RuntimeException(e);
    } finally { 
        try {
            if (fOut != null) {
                fOut.close();
            }
        } catch (IOException e) {
            throw new RuntimeException(e); 
        }
    } 
}

现在我们可以使用Schedulers.io()创建非阻塞的版本：

    Schedulers.io().createWorker().schedule(() -> {
        blockingStoreBitmap(context, bitmap, filename);
    }); 
}

每次我们调用storeBitmap()，RxJava处理创建所有它需要从I / O线程池一个特定的I/ O线程执行我们的任务。所有要执行的操作都避免在UI线程执行并且我们的App比之前要快上1秒：logcat上也不再有StrictMode的不合规做法。

SubscribeOn and ObserveOn

subscribeOn()方法来用于每个Observable对象。subscribeOn()方法用Scheduler来作为参数并在这个Scheduler上执行Observable调用

observeOn()方法将会在指定的调度器上返回结果

总结

　　RxJava是一个正在不断发展和扩大的世界。还有许多方法我们还没有去探索。有些方法甚至还没有，通过RxJava，你可以创建你自己的操作符并把他们发展地更远。+

　　Android是一个好玩的地方，但是它也有局限性。作为一个Android开发者，你可以用RxJava和RxAndroid克服其中的许多。我们用AndroidScheduler只简单提了下RxAndroid,除了在最后一章，你了解了ViewObservable。RxAndroid给了你许多：例如，WidgetObservable，LifecycleObservable。往后将它发展地更长远的任务就取决于你了。

谨记可观测序列就像一条河：它们是流动的。你可以“过滤”(filter)一条河，你可以“转换”(transform)一条河，你可以将两条河合并(combine)成一个，然后依然畅流如初。最后，它就成了你想要的那条河。

参考：http://www.aichengxu.com/view/2473291