rxjava源码线程切换（rxjava教程）

本文目录一览：

• 1、rxjava中doonsubscribe执行在哪个线程
• 2、rxjava怎么使用普通的
• 3、retrofit2+MVP+rxjava2+rxlifecycle2 为什么无法解决内存泄露
• 4、rxjava2 的生命周期绑定怎么和线程切换 结合
• 5、如何形象地描述RxJava中的背压和流控机制
• 6、rxjava flatmap中的操作可以在主线程中执行吗

rxjava中doonsubscribe执行在哪个线程

启动一个新的线程很简单，直接调用线程类的start()方法就可以了。但是必须是线程类，就是必须实现了Runnable接口的类，当然也可以是继承Thread类因为Thread也实现了Runnable接口。 Runnable接口有一个run()方法，java虚拟机会自己执行这个方法

rxjava怎么使用普通的

RxJava是 ReactiveX 在 Java 上的开源的实现。RxJava可以轻松处理不同运行环境下的后台线程或UI线程任务的框架。RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。

Observable（可观察者，即被观察者） 和 Subscriber（订阅者）是两个主要的类。在 RxJava 上，一个 Observable 是一个发出数据流或者事件的类，Subscriber 是一个对这些发出的 items （数据流或者事件）进行处理（采取行动）的类。

Observable 和Observer 通过 subscribe() 方法实现订阅关系。一个 Observable 的标准流发出一个或多个item，然后成功完成或者出错。一个 Observable 可以有多个 Subscribers，并且通过 Observable 发出的每一个 item，该 item 将会被发送到 Subscriber.onNext() 方法来进行处理。一旦 Observable 不再发出 items，它将会调用 Subscriber.onCompleted() 方法，或如果有一个出错的话Observable 会调用 Subscriber.onError() 方法。

onNext()： RxJava的事件回调方法，针对普通事件。

onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。

onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。

在一个正确运行的事件序列中,onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

图1-RxJava 的观察者模式（来自《给 Android 开发者的 RxJava 详解》）

RxAndroid是RxJava针对android平台增加的移植版本；使用RxAndroid也需要引入RxJava包。

retrofit2+MVP+rxjava2+rxlifecycle2 为什么无法解决内存泄露

随着Android第三库的普及， RxJava 和 RxAndroid ( tiveX/RxAndroid )越来越被人熟知，简洁的语法，配合Java8 Lambda表达式，使代码的结构更加清晰，通过线程调度器更容易控制和切换线程，种种优点，使用它的人也越来越多。但是使用不好，很容易导致内存泄露。 Rxlifecycle (b.com/trello/RxLifecycle )就使被用来严格控制由于发布了一个订阅后，由于没有及时取消，导致Activity/Fragment无法销毁导致的内存泄露 。

RxJava和RxAndroid

compile 'io.reactivex:rxandroid:1.0.1'compile 'io.reactivex:rxjava:1.0.16'12

Rxlifecycle

compile 'com.trello:rxlifecycle:0.3.0'

compile 'com.trello:rxlifecycle-components:0.3.0'12

Rxlifecycle 使用

Activity/Fragment需继承RxAppCompatActivity/RxFragment，目前支持的有RxAppCompatActivity、RxFragment、RxDialogFragment、RxFragmentActivity。

一、bindToLifecycle()方法

在子类使用Observable中的compose操作符，调用，完成Observable发布的事件和当前的组件绑定，实现生命周期同步。从而实现当前组件生命周期结束时，自动取消对Observable订阅。

Observable.interval(1, TimeUnit.SECONDS)

.compose(this.bindToLifecycle())

.subscribe(new Action1Long() {

@Override

public void call(Long num) {

Log.i(TAG, " " +num);

}

});12345678

二、bindUntilEvent() 方法

使用ActivityEvent类，其中的CREATE、START、 RESUME、PAUSE、STOP、 DESTROY分别对应生命周期内的方法。使用bindUntilEvent指定在哪个生命周期方法调用时取消订阅。

Observable.interval(1, TimeUnit.SECONDS) .compose(this.bindUntilEvent(ActivityEvent.PAUSE))

.subscribe(mSub);

rxjava2 的生命周期绑定怎么和线程切换 结合

二者区别如下：

1、RxJava有大量丰富强大的operator，可以满足用户的大部分数据处理需求。RxJava另一个强大的地方就是scheduler，用户可以为Observable和Subscriber指定不同的执行线程，在Android中可以方便的将Observable指定在IO线程中运行，Subscriber在UI线程中运行。

2、EventBus比较适合仅仅当做组件间的通讯工具使用，主要用来传递消息。使用EventBus可以避免搞出一大推的interface，仅仅是为了实现组件间的通讯，而不得不去实现那一推的接口。

联系：RxJava和EventBus一样也是基于观察者模式，但是使用的场景确实异步数据流的处理。

如何形象地描述RxJava中的背压和流控机制

在RxJava中，可以通过对Observable连续调用多个Operator组成一个调用链，其中数据从上游向下游传递。当上游发送数据的速度大于下游处理数据的速度时，就需要进行Flow Control了。

这就像小学做的那道数学题：一个水池，有一个进水管和一个出水管。如果进水管水流更大，过一段时间水池就会满（溢出）。这就是没有Flow Control导致的结果。

Flow Control有哪些思路呢？大概是有四种:

(1) 背压（Backpressure）。

(2) 节流（Throttling）。

(3) 打包处理。

(4) 调用栈阻塞（Callstack blocking）。

下面分别详细介绍。

注意：目前RxJava的1.x和2.x两个版本序列同时并存，2.x相对于1.x在接口上有很大变动，其中也包括Backpressure的部分。但是，这里要讨论的Flow Control机制中的相关概念，却都是适用的。

Flow Control的几种思路

背压（Backpressure）

Backpressure，也称为Reactive Pull，就是下游需要多少（具体是通过下游的request请求指定需要多少），上游就发送多少。这有点类似于TCP里的流量控制，接收方根据自己的接收窗口的情况来控制接收速率，并通过反向的ACK包来控制发送方的发送速率。

这种方案只对于所谓的cold Observable有效。cold Observable指的是那些允许降低速率的发送源，比如两台机器传一个文件，速率可大可小，即使降低到每秒几个字节，只要时间足够长，还是能够完成的。相反的例子是音视频直播，数据速率低于某个值整个功能就没法用了（这种就属于hot Observable了）。

节流（Throttling）

节流（Throttling），说白了就是丢弃。消费不过来，就处理其中一部分，剩下的丢弃。还是举音视频直播的例子，在下游处理不过来的时候，就需要丢弃数据包。

而至于处理哪些和丢弃哪些数据，就有不同的策略。主要有三种策略：

sample (也叫throttleLast)

throttleFirst

debounce (也叫throttleWithTimeout)

从细的方面分别解释一下。

sample，采样。类比一下音频采样，8kHz的音频就是每125微秒采一个值。sample可以配置成，比如每100毫秒采样一个值，但100毫秒内上游可能过来很多值，选哪个值呢，就是选最后那个值。所以它也叫throttleLast。

throttleFirst跟sample类似，比如还是每100毫秒采样一个值，但选这100毫秒内的第一个值。在Android开发中有时候可以把throttleFirst用作点击事件的防抖动处理，就是因为它可以在指定的一段时间内处理第一个点击事件（即采样第一个值），但丢弃后面的点击事件。

debounce，也叫throttleWithTimeout，名字里就包含一个例子。比如，一个网络程序维护一个TCP连接，不停地收发数据，但中间没数据可以收发的时候，就有间歇。这段间歇的时间，可以称为idle time。当idle time超过一个预设值的时候，就算超时了（time out），这个时候可能就需要把连接断开了。实际上一些做server端的网络程序就是这么工作的。每收发一个数据包之后，启动一个计时器，等待一个idle time。如果计时器到时之前，又有收发数据包的行为，那么计时器重置，等待一个新的idle time；而如果计时器时间到了，就超时了（time out），这个连接就可以关闭了。debounce的行为，跟这个非常类似，可以用它来找到那些连续的收发事件之后的idle time超时事件。换句话说，debounce可以把连续发生的事件之间的较大的间歇找出来。

打包处理

打包就是把上游来的小包裹打成大包裹，分发到下游。这样下游需要处理的包裹的个数就减少了。RxJava中提供了两类这样的机制：buffer和window。

buffer和window的功能基本一样，只是输出格式不太一样：buffer打包后的包裹用一个List表示，而window打包后的包裹又是一个Observable。

调用栈阻塞（Callstack blocking）

这是一种特殊情况，阻塞住整个调用栈（Callstack blocking）。之所以说这是一种特殊情况，是因为这种方式只适用于整个调用链都在一个线程上同步执行的情况，这要求中间的各个operator都不能启动新的线程。在平常使用中这种应该是比较少见的，因为我们经常使用subscribeOn或observeOn来切换执行线程，而且有些复杂的operator本身也会在内部启动新的线程来处理。另外，如果真的出现了完全同步的调用链，前面的另外三种Flow Control思路仍然可能是适用的，只不过这种阻塞的方式更简单，不需要额外的支持。

这里举个例子把调用栈阻塞和前面的Backpressure比较一下。“调用栈阻塞”相当于很多车行驶在盘山公路上，而公路只有一条车道。那么排在最前面的第一辆车就挡住了整条路，后面的车也只能排在后面。而“Backpressure”相当于银行办业务时的窗口叫号，窗口主动叫某个号过去（相当于请求），那个人才过去办理。

如何让Observable支持Backpressure？

在RxJava 1.x中，有些Observable是支持Backpressure的，而有些不支持。但不支持Backpressure的Observable可以通过一些operator来转化成支持Backpressure的Observable。这些operator包括：

onBackpressureBuffer

onBackpressureDrop

onBackpressureLatest

onBackpressureBlock（已过期）

它们转化成的Observable分别具有不同的Backpressure策略。

而在RxJava 2.x中，Observable不再支持Backpressure，而是改用Flowable来专门支持Backpressure。上面提到的四种operator的前三种分别对应Flowable的三种Backpressure策略：

BackpressureStrategy.BUFFER

BackpressureStrategy.DROP

BackpressureStrategy.LATEST

onBackpressureBuffer是不丢弃数据的处理方式。把上游收到的全部缓存下来，等下游来请求再发给下游。相当于一个水库。但上游太快，水库（buffer）就会溢出。

onBackpressureDrop和onBackpressureLatest比较类似，都会丢弃数据。这两种策略相当于一种令牌机制（或者配额机制），下游通过request请求产生令牌（配额）给上游，上游接到多少令牌，就给下游发送多少数据。当令牌数消耗到0的时候，上游开始丢弃数据。但这两种策略在令牌数为0的时候有一点微妙的区别：onBackpressureDrop直接丢弃数据，不缓存任何数据；而onBackpressureLatest则缓存最新的一条数据，这样当上游接到新令牌的时候，它就先把缓存的上一条“最新”数据发送给下游。可以结合下面两幅图来理解。

onBackpressureBlock是看下游有没有需求，有需求就发给下游，下游没有需求，不丢弃，但试图堵住上游的入口（能不能真堵得住还得看上游的情况了），自己并不缓存。这种策略已经废弃不用。

rxjava flatmap中的操作可以在主线程中执行吗

1.去官网大致看一遍各个操作符2.部分区分不清（比如map与flatmap）的面向github或者stackoverflow找答案，用英文哦，否则会被百度与CSDN的渣文章给扰乱3.去github等社区找android示例，比如GitHub-lzyzsd/Awesome-RxJava:RxJavaresources4.自己去写实际项目或者编程题，比如去oj平台（比如newcoder）用rxjava做编程题，这个非常有助于加深对流程化与可观察的理解。虽然平台上不可能用第三方lib，但是自己只用rxjava写出来并与原来算法进行对比（代码量与可读性）即可，最后整理成笔记博客等进行保存。