Carson带你学Android：RxJava线程控制(含实例讲解)

淡淡的烟草味﹌ 2022-09-11 07:18 218阅读 0赞

![a26aedafc2ac45510787091b1d02f019.png][]

--------------------

# 前言 #

`Rxjava`由于其**基于事件流的链式调用、逻辑简洁 & 使用简单**的特点，深受各大 `Android`开发者的欢迎。本文主要讲解的是： **线程控制（也称为调度 / 切换）**，即讲解功能性操作符中的：`subscribeOn（） & observeOn（）`

> Carson带你学RxJava系列文章，包括 **原理、操作符、应用场景、背压等等**，请看文章：[Android：这是一份全面 & 详细的RxJava学习指南][Android_ _ _RxJava]

![7ff71db5180886f1cd42bb3e7035ab5a.png][]

--------------------

# 目录 #

![示意图][f9ecee82265bfe960d358bdc7d572442.png]

--------------------

# 1. RxJava线程控制（调度 / 切换）的作用是什么？ #

指定 被观察者 `（Observable）` / 观察者`（Observer）` 的工作线程类型。

--------------------

# 2. 为什么要进行RxJava线程控制（调度 / 切换）？ #

### 2.1 背景 ###

*  在 `RxJava`模型中，**被观察者 `（Observable）` / 观察者`（Observer）`的工作线程 = 创建自身的线程**

> 即，若被观察者 `（Observable）` / 观察者`（Observer）`在主线程被创建，那么他们的工作（生产事件 / 接收& 响应事件）就会发生在主线程

*  因为创建被观察者 `（Observable）` / 观察者`（Observer）`的线程 = 主线程
 *  所以生产事件 / 接收& 响应事件都发生在主线程

> 下面请看1个RxJava的基础使用

public class MainActivity extends AppCompatActivity {
    
        private static final String TAG = "Rxjava";
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);
    
            // 步骤1：创建被观察者 Observable & 发送事件
            // 在主线程创建被观察者 Observable 对象
            // 所以生产事件的线程是：主线程
    
            Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
    
                    Log.d(TAG, " 被观察者 Observable的工作线程是: " + Thread.currentThread().getName());
                    // 打印验证
                    emitter.onNext(1);
                    emitter.onComplete();
                }
            });
    
    // 步骤2：创建观察者 Observer 并 定义响应事件行为
            // 在主线程创建观察者 Observer 对象
            // 所以接收 & 响应事件的线程是：主线程
            Observer<Integer> observer = new Observer<Integer>() {
    
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "开始采用subscribe连接");
                    Log.d(TAG, " 观察者 Observer的工作线程是: " + Thread.currentThread().getName());
                    // 打印验证
    
                }
                @Override
                public void onNext(Integer value) {
                    Log.d(TAG, "对Next事件"+ value +"作出响应"  );
                }
                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "对Error事件作出响应");
                }
                @Override
                public void onComplete() {
                    Log.d(TAG, "对Complete事件作出响应");
                }
            };
    
            // 步骤3：通过订阅（subscribe）连接观察者和被观察者
            observable.subscribe(observer);
        }
    }

*  测试结果

![示意图][17861ef739d69a0a398067db24fad547.png]

### 2.2 冲突 ###

*  对于一般的需求场景，需要在子线程中实现耗时的操作；然后回到主线程实现 `UI`操作
 *  应用到 `RxJava`模型中，可理解为：
    
    1.  被观察者 `（Observable）` 在 **子线程** 中生产事件（如实现耗时操作等等）
    2.  观察者`（Observer）`在 **主线程** 接收 & 响应事件（即实现UI操作）

![示意图][4ab1f8bb309371ad5c36b508769c114e.png]

### 2.3 解决方案 ###

所以，为了解决上述冲突，即实现 **真正的异步操作**，我们需要对`RxJava`进行 **线程控制（也称为调度 / 切换）**

--------------------

# 3. 实现方式 #

采用 `RxJava`内置的**线程调度器**（ `Scheduler` ），即通过 \*\*功能性操作符`subscribeOn（）` & `observeOn（）`\*\*实现

### 3.1 功能性操作符subscribeOn（） & observeOn（）简介 ###

*  作用  
    线程控制，即指定 被观察者 `（Observable）` / 观察者`（Observer）` 的工作线程类型
 *  线程类型  
    在 `RxJava`中，内置了多种用于调度的线程类型

<table> 
 <thead> 
  <tr> 
   <th>类型</th> 
   <th align="center">含义</th> 
   <th align="right">应用场景</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Schedulers.immediate()</td> 
   <td align="center">当前线程 = 不指定线程</td> 
   <td align="right">默认</td> 
  </tr> 
  <tr> 
   <td>AndroidSchedulers.mainThread()</td> 
   <td align="center">Android主线程</td> 
   <td align="right">操作UI</td> 
  </tr> 
  <tr> 
   <td>Schedulers.newThread()</td> 
   <td align="center">常规新线程</td> 
   <td align="right">耗时等操作</td> 
  </tr> 
  <tr> 
   <td>Schedulers.io()</td> 
   <td align="center">io操作线程</td> 
   <td align="right">网络请求、读写文件等io密集型操作</td> 
  </tr> 
  <tr> 
   <td>Schedulers.computation()</td> 
   <td align="center">CPU计算操作线程</td> 
   <td align="right">大量计算操作</td> 
  </tr> 
 </tbody> 
</table>

*  注：`RxJava`内部使用 **线程池** 来维护这些线程，所以线程的调度效率非常高。

### 3.2 具体使用 ###

*  具体是在 （上述步骤3）**通过订阅（subscribe）连接观察者和被观察者**中实现

<-- 使用说明 -->
      // Observable.subscribeOn（Schedulers.Thread）：指定被观察者 发送事件的线程（传入RxJava内置的线程类型）
      // Observable.observeOn（Schedulers.Thread）：指定观察者 接收 & 响应事件的线程（传入RxJava内置的线程类型）
    
    <-- 实例使用 -->
    // 步骤3：通过订阅（subscribe）连接观察者和被观察者
            observable.subscribeOn(Schedulers.newThread()) // 1. 指定被观察者 生产事件的线程
                      .observeOn(AndroidSchedulers.mainThread())  // 2. 指定观察者 接收 & 响应事件的线程
                      .subscribe(observer); // 3. 最后再通过订阅（subscribe）连接观察者和被观察者

*  测试结果  
    ![示意图][6b35fe6cf29121604d7076a16c9d9472.png]
 *  特别注意

###### 1. 若`Observable.subscribeOn（）`多次指定被观察者 生产事件的线程，则只有第一次指定有效，其余的指定线程无效 ######

// 步骤3：通过订阅（subscribe）连接观察者和被观察者
            observable.subscribeOn(Schedulers.newThread()) // 第一次指定被观察者线程 = 新线程
                      .subscribeOn(AndroidSchedulers.mainThread()) // 第二次指定被观察者线程 = 主线程
                      .observeOn(AndroidSchedulers.mainThread())
                      .subscribe(observer);

*  测试结果：被观察者的线程 = 第一次指定的线程 = 新的工作线程，第二次指定的线程（主线程）无效

![示意图][c879e7a9e2a8e394ee0711821da86e23.png]

###### 2. 若Observable.observeOn（）多次指定观察者 接收 & 响应事件的线程，则每次指定均有效，即每指定一次，就会进行一次线程的切换 ######

// 步骤3：通过订阅（subscribe）连接观察者和被观察者
            observable.subscribeOn(Schedulers.newThread())
                      .observeOn(AndroidSchedulers.mainThread()) // 第一次指定观察者线程 = 主线程
                      .doOnNext(new Consumer<Integer>() { // 生产事件
                        @Override
                        public void accept(Integer integer) throws Exception {
                            Log.d(TAG, "第一次观察者Observer的工作线程是： " + Thread.currentThread().getName());
                        }
                    })
                    .observeOn(Schedulers.newThread()) // 第二次指定观察者线程 = 新的工作线程
                    .subscribe(observer); // 生产事件
    
    
    // 注：
    // 1. 整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.doOnNext（）>观察者.onNext（）>观察者.onComplete() 
    // 2. 观察者.onSubscribe（）固定在主线程进行

*  测试结果：每调用一次`observeOn()`，观察者的线程就会切换一次

![示意图][0a59db012e173dddd84dd0efb2cc7dfd.png]

--------------------

# 4. 具体实例 #

下面，我将采用最常见的 `Retrofit + RxJava` 实现 网络请求 的功能，从而说明 `RxJava`的线程控制的具体应用

### 4.1 功能说明 ###

*  实现功能：将中文翻译成英文 - > 显示到界面
 *  实现方案：采用`Get`方法对 金山词霸API 发送网络请求

> 1.  先切换到工作线程 发送网络请求
> 2.  再切换到主线程进行 `UI`更新

![金山词典][ac52b3dc0a0eeda4bef78c0766c453cb.png]

### 4.2 步骤说明 ###

1.  添加依赖
2.  创建 接收服务器返回数据 的类
3.  创建 用于描述网络请求 的接口（区别于传统形式）
4.  创建 Retrofit 实例
5.  创建 网络请求接口实例 并 配置网络请求参数（区别于传统形式）
6.  发送网络请求（区别于传统形式）
7.  发送网络请求
8.  对返回的数据进行处理

> 本实例侧重于说明 `RxJava` 的线程控制，关于`Retrofit`的使用请看文章：[这是一份很详细的 Retrofit 2.0 使用教程（含实例讲解）][Retrofit 2.0]

### 4.3 步骤实现 ###

#### 步骤1： 添加依赖 ####

a. 在 `Gradle`加入`Retrofit`库的依赖

*build.gradle*

dependencies {
    
    // Android 支持 Rxjava
    // 此处一定要注意使用RxJava2的版本
    compile 'io.reactivex.rxjava2:rxjava:2.0.1'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
    
    // Android 支持 Retrofit
    compile 'com.squareup.retrofit2:retrofit:2.1.0'
    
    // 衔接 Retrofit & RxJava
    // 此处一定要注意使用RxJava2的版本
    compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
    
    // 支持Gson解析
    compile 'com.squareup.retrofit2:converter-gson:2.1.0'
    
    }

b. 添加 网络权限  
*AndroidManifest.xml*

<uses-permission android:name="android.permission.INTERNET"/>

##### 步骤2：创建 接收服务器返回数据 的类 #####

*  金山词霸`API` 的数据格式说明如下：

// URL模板
    http://fy.iciba.com/ajax.php
    
    // URL实例
    http://fy.iciba.com/ajax.php?a=fy&f=auto&t=auto&w=hello%20world
    
    // 参数说明：
    // a：固定值 fy
    // f：原文内容类型，日语取 ja，中文取 zh，英语取 en，韩语取 ko，德语取 de，西班牙语取 es，法语取 fr，自动则取 auto
    // t：译文内容类型，日语取 ja，中文取 zh，英语取 en，韩语取 ko，德语取 de，西班牙语取 es，法语取 fr，自动则取 auto
    // w：查询内容

*  示例

![API格式说明][API]

*  根据 金山词霸API 的数据格式，创建 接收服务器返回数据 的类：

*Translation.java*

public class Translation {
        private int status;
    
        private content content;
        private static class content {
            private String from;
            private String to;
            private String vendor;
            private String out;
            private int errNo;
        }
    
        //定义 输出返回数据 的方法
        public void show() {
            System.out.println( "Rxjava翻译结果：" + status);
            System.out.println("Rxjava翻译结果：" + content.from);
            System.out.println("Rxjava翻译结果：" + content.to);
            System.out.println("Rxjava翻译结果：" + content.vendor);
            System.out.println("Rxjava翻译结果：" + content.out);
            System.out.println("Rxjava翻译结果：" + content.errNo);
        }
    }

##### 步骤3：创建 用于描述网络请求 的接口 #####

采用 **注解** \+ `Observable<...>`接口描述 网络请求参数

*GetRequest\_Interface.java*

public interface GetRequest_Interface {
    
        @GET("ajax.php?a=fy&f=auto&t=auto&w=hi%20world")
        Observable<Translation> getCall();
         // 注解里传入 网络请求 的部分URL地址
        // Retrofit把网络请求的URL分成了两部分：一部分放在Retrofit对象里，另一部分放在网络请求接口里
        // 如果接口里的url是一个完整的网址，那么放在Retrofit对象里的URL可以忽略
        // 采用Observable<...>接口 
        // getCall()是接受网络请求数据的方法
    }

##### 接下来的步骤均在*MainActivity.java*内实现（请看注释） #####

*MainActivity.java*

public class MainActivity extends AppCompatActivity {
    
        private static final String TAG = "Rxjava";
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);
    
            //步骤4：创建Retrofit对象
            Retrofit retrofit = new Retrofit.Builder()
                    .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                    .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                    .build();
    
            // 步骤5：创建 网络请求接口 的实例
            GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
    
            // 步骤6：采用Observable<...>形式 对 网络请求 进行封装
            Observable<Translation> observable = request.getCall();
    
            // 步骤7：发送网络请求
            observable.subscribeOn(Schedulers.io())               // 在IO线程进行网络请求
                      .observeOn(AndroidSchedulers.mainThread())  // 回到主线程 处理请求结果
                      .subscribe(new Observer<Translation>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.d(TAG, "开始采用subscribe连接");
                        }
    
                        @Override
                        public void onNext(Translation result) {
                            // 步骤8：对返回的数据进行处理
                            result.show() ;
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            Log.d(TAG, "请求失败");
                        }
    
                        @Override
                        public void onComplete() {
                            Log.d(TAG, "请求成功");
                        }
                    });
        }
    }

### 4.4 测试结果 ###

![示意图][8ea2390db177c9a19f2e6cffe092a71d.png]

### 4.5 Demo地址 ###

[Carson\_Ho的Github地址 = RxJava2实战系列：线程控制][Carson_Ho_Github_ _ RxJava2]

--------------------

# 5. 注意事项 #

### 5.1 依赖包问题 ###

*  问题说明

![示意图][0c038a5ebe51e5ed394c1abf030b6171.png]

*  解决方法  
    通过在`Gradle`使用`packageOptions`解决

*build.gradle*

android {
        ...
        packagingOptions {
            exclude 'META-INF/rxjava.properties'
        }
    }

### 5.2 应用程序崩溃问题 ###

*  背景：在发送网络请求时 退出当前`Activity`
 *  冲突：此时如果回到主线程更新 `UI`，`App`会崩溃
 *  解决方案：当 `Activity`退出时，调用 `Disposable.dispose()`切断观察者和被观察者的连接，使得观察者无法收到事件 & 响应事件

> 当出现多个`Disposable`时，可采用`RxJava`内置容器`CompositeDisposable`进行统一管理

// 添加Disposable到CompositeDisposable容器
    CompositeDisposable.add()
    
    // 清空CompositeDisposable容器
    CompositeDisposable.clear()

--------------------

# 6. 总结 #

*  本文主要对 `Rxjava` 中的线程调度、功能性操作符`subscribeOn()` & `observeOn()`进行讲解
 *  Carson带你学RxJava系列文章：

> 入门  
> [Carson带你学Android：这是一篇清晰易懂的Rxjava入门教程][Carson_Android_Rxjava]  
> [Carson带你学Android：这是一份面向初学者的RxJava使用指南][Carson_Android_RxJava]  
> [Carson带你学Android：RxJava2.0到底更新了什么？(含使用建议)][Carson_Android_RxJava2.0]  
> 原理  
> [Carson带你学Android：图文解析带你快速了解RxJava原理][Carson_Android_RxJava 1]  
> [Carson带你学Android：手把手带你源码分析RxJava][Carson_Android_RxJava 2]  
> 使用教程(操作符)  
> [Carson带你学Android：RxJava操作符教程][Carson_Android_RxJava 3]  
> [Carson带你学Android：RxJava创建操作符][Carson_Android_RxJava 4]  
> [Carson带你学Android：RxJava功能性操作符][Carson_Android_RxJava 5]  
> [Carson带你学Android：RxJava过滤操作符][Carson_Android_RxJava 6]  
> [Carson带你学Android：RxJava组合/合并操作符][Carson_Android_RxJava 7]  
> [Carson带你学Android：RxJava变换操作符][Carson_Android_RxJava 8]  
> [Carson带你学Android：RxJava条件/布尔操作符][Carson_Android_RxJava 9]  
> 应用  
> [Carson带你学Android：什么时候应该使用Rxjava？(开发场景汇总)][Carson_Android_Rxjava 1]  
> [Carson带你学Android：RxJava线程控制(含实例讲解)][Carson_Android_RxJava 10]  
> [Carson带你学Android：图文详解RxJava背压策略][Carson_Android_RxJava 11]  
> [Carson带你学Android：RxJava、Retrofit联合使用汇总(含实例教程)][Carson_Android_RxJava_Retrofit]  
> [Carson带你学Android：优雅实现网络请求嵌套回调][Carson_Android]  
> [Carson带你学Android：网络请求轮询(有条件)][Carson_Android 1]  
> [Carson带你学Android：网络请求轮询(无条件)][Carson_Android 2]  
> [Carson带你学Android：网络请求出错重连(结合Retrofit)][Carson_Android_Retrofit]  
> [Carson带你学Android：合并数据源][Carson_Android 3]  
> [Carson带你学Android：联想搜索优化][Carson_Android 4]  
> [Carson带你学Android：功能防抖][Carson_Android 5]  
> [Carson带你学Android：从磁盘/内存缓存中获取缓存数据][Carson_Android 6]  
> [Carson带你学Android：联合判断][Carson_Android 7]

--------------------

# 欢迎关注Carson\_Ho的[CSDN博客][CSDN] #

博客链接：https://carsonho.blog.csdn.net/  
![a26aedafc2ac45510787091b1d02f019.png][]

--------------------

# 请帮顶 / 评论点赞！因为你的鼓励是我写作的最大动力！ #

[a26aedafc2ac45510787091b1d02f019.png]: /images/20220828/da5c0702d0334f699fa0b715177893e7.png
[Android_ _ _RxJava]: https://blog.csdn.net/carson_ho/article/details/120326253
[7ff71db5180886f1cd42bb3e7035ab5a.png]: /images/20220828/65e881fb64484c86b54b0ae99d534404.png
[f9ecee82265bfe960d358bdc7d572442.png]: /images/20220828/0d69e3cb479c4d4993a1b4c9ae90eb63.png
[17861ef739d69a0a398067db24fad547.png]: /images/20220828/6e4d0cb1fa1946639b804f75ed8fb269.png
[4ab1f8bb309371ad5c36b508769c114e.png]: /images/20220828/52602398ed9848ed9115e88a2d11c6ff.png
[6b35fe6cf29121604d7076a16c9d9472.png]: /images/20220828/02ae927982c847fc98b939e5a3c4e31f.png
[c879e7a9e2a8e394ee0711821da86e23.png]: /images/20220828/30faa80a64834ea282d19d58ab25e74b.png
[0a59db012e173dddd84dd0efb2cc7dfd.png]: /images/20220828/e44ec4cb4dac491091c57da5f1bfe325.png
[ac52b3dc0a0eeda4bef78c0766c453cb.png]: /images/20220828/1889f74f64554772bf915e97e60ab66f.png
[Retrofit 2.0]: http://www.jianshu.com/p/a3e162261ab6
[API]: /images/20220828/95b43480981f4bd58a8d85071f33f232.png
[8ea2390db177c9a19f2e6cffe092a71d.png]: /images/20220828/d03d1cc18e0740348a9fdd8dd05fde51.png
[Carson_Ho_Github_ _ RxJava2]: https://github.com/Carson-Ho/RxJava_Operators
[0c038a5ebe51e5ed394c1abf030b6171.png]: /images/20220828/eaa50c3f6159422fb47fc5913c265647.png
[Carson_Android_Rxjava]: https://blog.csdn.net/carson_ho/article/details/78179340
[Carson_Android_RxJava]: https://blog.csdn.net/carson_ho/article/details/79168602
[Carson_Android_RxJava2.0]: https://blog.csdn.net/carson_ho/article/details/79157408
[Carson_Android_RxJava 1]: https://carsonho.blog.csdn.net/article/details/100112005
[Carson_Android_RxJava 2]: https://blog.csdn.net/carson_ho/article/details/79168523
[Carson_Android_RxJava 3]: https://blog.csdn.net/carson_ho/article/details/79191327
[Carson_Android_RxJava 4]: https://blog.csdn.net/carson_ho/article/details/78246732
[Carson_Android_RxJava 5]: https://www.jianshu.com/p/b0c3669affdb
[Carson_Android_RxJava 6]: https://blog.csdn.net/carson_ho/article/details/78683064
[Carson_Android_RxJava 7]: https://blog.csdn.net/carson_ho/article/details/78455349
[Carson_Android_RxJava 8]: https://blog.csdn.net/carson_ho/article/details/78315437
[Carson_Android_RxJava 9]: https://blog.csdn.net/carson_ho/article/details/78949306
[Carson_Android_Rxjava 1]: https://blog.csdn.net/carson_ho/article/details/79168723
[Carson_Android_RxJava 10]: https://carsonho.blog.csdn.net/article/details/120282158
[Carson_Android_RxJava 11]: https://blog.csdn.net/carson_ho/article/details/79081407
[Carson_Android_RxJava_Retrofit]: https://blog.csdn.net/carson_ho/article/details/79125101
[Carson_Android]: https://blog.csdn.net/carson_ho/article/details/78315696
[Carson_Android 1]: https://blog.csdn.net/carson_ho/article/details/78558790
[Carson_Android 2]: https://blog.csdn.net/carson_ho/article/details/78256466
[Carson_Android_Retrofit]: https://blog.csdn.net/carson_ho/article/details/78651602
[Carson_Android 3]: https://blog.csdn.net/carson_ho/article/details/78455544
[Carson_Android 4]: https://blog.csdn.net/carson_ho/article/details/78849661
[Carson_Android 5]: https://blog.csdn.net/carson_ho/article/details/78849689
[Carson_Android 6]: https://blog.csdn.net/carson_ho/article/details/78455449
[Carson_Android 7]: https://blog.csdn.net/carson_ho/article/details/78455624
[CSDN]: https://carsonho.blog.csdn.net/