Android RxJava实际应用案例讲解：使用RxJava的最佳开发场景

阅读量：6119 次

发布时间：2019-06-21

本文共 5175 字，大约阅读时间需要 17 分钟。

前言

Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 Android开发者的欢迎。

Github截图

如果还不了解RxJava，请看文章：

今天，我将为大家带来 Rxjava的的基本使用 & 实际应用案例教学，即常见开发应用场景实现，并结合常用相关框架如Retrofit等，希望大家会喜欢。
1. 本系列文章主要基于 Rxjava 2.0
2. 接下来的时间，我将持续推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操作符、应用场景、背压等等，有兴趣可以继续关注！！

1. 简介

RxJava的简介如下

示意图

2. 基本使用

Rxjava的使用方式有两种：
方式1：分步骤实现
方式2：基于事件流的链式调用

具体使用
请看文章

3. 实际开发应用场景

RxJava的实际开发应用场景与其对应的操作符息息相关

常见的RxJava实际开发应用场景有如下：

示意图

下面，我将对每个实际开发应用场景进行实例讲解教学

下面实例皆结合常用框架如
Retrofit 、
RxBinding、
RxBus等

3.1 网络请求轮询（无条件）

需求场景

具体实现

3.2 网路请求轮询（有条件）

需求场景

具体实现

3.3 网络请求出错重连

需求场景

功能需求说明

示意图

功能逻辑

示意图

具体实现

3.4 网络请求嵌套回调

背景
需要进行嵌套网络请求：即在第1个网络请求成功后，继续再进行一次网络请求

如先进行用户注册的网络请求, 待注册成功后回再继续发送用户登录的网络请求

冲突
嵌套实现网络请求较为复杂，即嵌套调用函数

下面展示的是结合 Retrofit 与 RxJava的基本用法，即未用操作符前

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

解决方案
结合 RxJava2中的变换操作符FlatMap（）实现嵌套网络请求

具体实现

3.5 从磁盘 / 内存缓存中获取缓存数据

需求场景

功能说明
对于从磁盘 / 内存缓存中获取缓存数据的功能逻辑如下：

具体实现

3.6 合并数据源

需求场景

示意图

功能说明
即，同时向2个数据源获取数据 -> 合并数据 -> 统一展示到客户端

具体实现

3.7 联合判断

需求场景
需要同时对多个事件进行联合判断

如，填写表单时，需要表单里所有信息（姓名、年龄、职业等）都被填写后，才允许点击 “提交” 按钮

功能说明
此处采用填写表单作为联合判断功能展示，即，表单里所有信息（姓名、年龄、职业等）都被填写后，才允许点击 “提交” 按钮

具体实现

3.8 线程控制（切换 / 调度）

需求场景
即，新开工作线程执行耗时操作；待执行完毕后，切换到主线程实时更新 UI

具体实现

3.9 功能防抖

需求场景

功能说明

示意图

具体实现

3.10 联想搜索优化

需求场景

功能说明

示意图

具体实现

3.11 控制被观察者发送事件 & 观察者接收事件速度：背压

a. 背景

观察者 & 被观察者之间存在2种订阅关系：同步 & 异步。具体如下：

示意图

对于异步订阅关系，存在被观察者发送事件速度与观察者接收事件速度不匹配的情况
1. 发送 & 接收事件速度 = 单位时间内发送&接收事件的数量
2. 大多数情况，主要是被观察者发送事件速度＞观察者接收事件速度

b. 冲突

被观察者发送事件速度太快，而观察者来不及接收所有事件，从而导致观察者无法及时响应 / 处理所有发送过来事件的问题，最终导致缓存区溢出、事件丢失 & OOM

如，点击按钮事件：连续过快的点击按钮10次，则只会造成点击2次的效果；

解释：因为点击速度太快了，所以按钮来不及响应

下面再举个例子：

被观察者的发送事件速度 = 10ms / 个

观察者的接收事件速度 = 5s / 个

即出现发送 & 接收事件严重不匹配的问题

Observable.create(new ObservableOnSubscribe
      
       () {            // 1. 创建被观察者 & 生产事件            @Override            public void subscribe(ObservableEmitter
       
         emitter) throws Exception {                for (int i = 0; ; i++) {                    Log.d(TAG, "发送了事件"+ i );                    Thread.sleep(10);                    // 发送事件速度：10ms / 个                     emitter.onNext(i);                }            }        }).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行                .observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行             .subscribe(new Observer
        
         () {            // 2. 通过通过订阅（subscribe）连接观察者和被观察者            @Override            public void onSubscribe(Disposable d) {                Log.d(TAG, "开始采用subscribe连接");            }            @Override            public void onNext(Integer value) {                try {                    // 接收事件速度：5s / 个                     Thread.sleep(5000);                    Log.d(TAG, "接收到了事件"+ value  );                } catch (InterruptedException e) {                    e.printStackTrace();                }            }            @Override            public void onError(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }            @Override            public void onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        });