【Rxjava详解】(二) 操作符的妙用

文章目录

- 接口变化
- 操作符
- - map
  - flatmap
  - debounce
  - throttleFirst()
  - take
  - concat

RxJava 是一个基于 观察者模式的异步编程库，它提供了丰富的操作符来处理和转换数据流。操作符是 RxJava 的核心组成部分，它们提供了一种灵活、可组合的方式来处理数据流，使得开发者可以更加便捷地进行数据处理和流程控制。

接口变化

RxJava 2.x拥有了新的特性，其依赖于4个接口:

Publisher
Subscriber
Subscription
Processor

为Subscriber，Func1变为Function等等。此外，还引入了Single、Maybe和Completable等新的可观察类型。
RxJava 2.x 中的背压支持：RxJava 2.x 引入了对背压的支持，新增了Flowable类型来处理背压场景。同时，对一些操作符的行为进行了一些修改以适应背压机制。
异常处理方式的变化：在RxJava 1.x中，异常处理是通过onError()方法来处理，而在RxJava 2.x中，引入了onError(Throwable)方法和onError(Throwable, boolean)方法，允许开发者控制是否中断流程。
取消订阅的方式变化：在RxJava 1.x中，使用unsubscribe()方法取消订阅，而在RxJava 2.x中，使用dispose()方法取消订阅

关于背压：

在RxJava中，背压（Backpressure）是一种处理生产者和消费者之间速度不匹配的机制。通过背压，可以使得消费者根据自身的处理能力告知生产者它们能够接受的数据量，从而避免生产者产生过多的数据导致消费者无法处理的情况。

而在RxJava 2.x ~ RxJava 3.x，发生以下变化：

不再支持Backpressure：RxJava 3.x不再内置支持背压机制，而是采用基于Reactive-Streams的响应式规范，并提供了相应的Flowable类型。因此，在RxJava 3.x中，需要使用Flowable来处理背压场景。
Observer接口的变化：在RxJava 3.x中，Observer接口被拆分为两个接口：Observer和Disposable。Observer接口用于处理事件的消费，而Disposable接口用于取消订阅。
SingleObserver和CompletableObserver的变化：在RxJava 3.x中，SingleObserver和CompletableObserver接口的方法签名有所变化，取消订阅的方法从dispose()改为了onDispose()。

操作符

RxJava提供了对事件序列进行变换的支持，这是它的核心功能之一.所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。
操作符是用来对Observable（或Flowable）流进行转换、过滤、组合和操作的方法。

RxJava提供了很多很有用的操作符。~~多的要死~~
在Rxjava 3.x 下，有以下常见的操作符：

map：将Observable发射的数据项通过指定的函数进行转换，并发射转换后的数据项。
filter：根据指定的条件过滤Observable发射的数据项，只发射满足条件的数据项。
take：只发射Observable发射的前N个数据项，忽略后面的数据项。
skip：跳过Observable发射的前N个数据项，只发射后面的数据项。
merge：将多个Observable合并成一个Observable，按照时间顺序发射合并后的数据项。
zip：将多个Observable按照顺序进行合并，每个数据项都是由对应位置的Observable发射的数据项组合而成。
concat：按照顺序连接多个Observable，依次发射它们的数据项，等前一个Observable完成后才会订阅下一个Observable。
onErrorResumeNext：在Observable发生错误时，使用备用的Observable继续发射数据项。
retry：在Observable发生错误时，进行错误重试，重新订阅Observable。
interval：创建一个按照固定时间间隔发射递增数值的Observable。
debounce：只有在指定的时间间隔内没有发射新的数据项时，才发射最后一个数据项。
distinct：过滤掉重复的数据项，只发射不重复的数据项。
flatMap：将Observable发射的数据项转换为Observable集合，并按顺序发射这些Observable发射的数据项。
reduce：对Observable发射的数据项进行累积操作，返回最终的累积结果。
scan：对Observable发射的数据项进行累积操作，并按顺序发射每次累积的结果

在此简单介绍其中几个的用法：

map

示意图：

实际上，map操作符可以理解为对Observable发射的每个数据项都应用一个函数，将原始数据项转换为另一种形式的数据项，然后再发射出去。（感觉Kotlin里有）

假设我们有一个Observable发射的是整数序列，我们想将每个整数乘以2，并发射出去。

Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5);

observable
    .map(number -> number * 2)
    .subscribe(result -> System.out.println(result));

输出：
2
4
6
8
10

flatmap

flatMap操作符会对Observable的每个数据项应用一个函数，这个函数返回一个新的Observable。然后，它会将这些新的Observable合并成一个Observable，并发射合并后的数据项。

假设我们有一个Observable发射的是字符串数组，我们想将每个字符串拆分为字符数组，并发射出去。

Observable<String> observable = Observable.just("Hello", "World", "RxJava");

observable
    .flatMap(str -> Observable.fromArray(str.split("")))
    .subscribe(character -> System.out.print(character + " "));

输出：

H e l l o W o r l d R x J a v a

flatMap操作符将每个字符串拆分为字符数组，并将所有的字符合并成了一个Observable，最终发射出去。

debounce

debounce操作符也是RxJava中常用的操作符之一，它用于在一定时间间隔内只发射最后一个数据项，忽略中间的数据项。debounce操作符主要用于处理需要在一定时间内连续发生的事件，但只关心最后一个事件的场景。

在安卓开发中，debounce操作符可以用于处理用户输入场景，比如搜索框输入关键词时，通常需要等待用户停止输入一段时间后再进行搜索，以减少不必要的网络请求。

Observable<String> observable = Observable.create(emitter -> {
    editText.addTextChangedListener(new TextWatcher() {
        @Override
        public void beforeTextChanged(CharSequence s, int start, int count, int after) {
        }

        @Override
        public void onTextChanged(CharSequence s, int start, int before, int count) {
        }

        @Override
        public void afterTextChanged(Editable s) {
            emitter.onNext(s.toString());
        }
    });
});

observable
    .debounce(500, TimeUnit.MILLISECONDS)
    .subscribe(keyword -> {
        // 进行搜索操作
        performSearch(keyword);
    });

我们首先创建了一个Observable对象，该Observable通过监听EditText的文本变化事件，将用户输入的关键词发射出去。

然后，我们使用debounce操作符，设置一个时间间隔（这里是500毫秒），它会在用户输入停止500毫秒后才发射最后一个关键词。

最后，通过subscribe方法订阅Observable，并在订阅中执行搜索操作。

这样做的好处是，用户在连续输入时，debounce操作符会忽略中间的输入，只关注最后一个输入，在用户停止输入一段时间后才执行搜索操作，避免不必要的网络请求。

throttleFirst()

throttleFirst()操作符也是RxJava中常用的操作符之一，它用于在指定时间间隔内只发射第一个数据项，忽略后续的数据项。throttleFirst操作符主要用于处理需要限制触发频率的事件，保证在指定时间间隔内只处理一次。

在安卓开发中，throttleFirst操作符可以用于处理按钮点击事件，防止用户重复点击按钮造成重复操作：

Observable<Object> observable = Observable.create(emitter -> {
    button.setOnClickListener(v -> {
        emitter.onNext(new Object());
    });
});

observable
    .throttleFirst(1000, TimeUnit.MILLISECONDS)
    .subscribe(event -> {
        // 执行按钮点击操作
        ClickAction();
    });

take

take()操作符也是RxJava中常用的操作符之一，用于从Observable中取出一定数量的数据项，并在达到指定数量后停止发射。它可以与Retrofit和RxJava的线程切换一起使用，来控制网络请求结果的数量和线程切换。

在安卓开发中，通常使用Retrofit进行网络请求，而结合RxJava可以实现异步操作和线程切换。下面是一个结合Retrofit和RxJava的实例，使用take操作符来限制结果数量，并配合线程切换：

首先，创建一个网络请求的接口：

public interface ApiInterface {
    @GET("data")
    Observable<List<Data>> getData();
}

然后，创建一个Retrofit实例，并结合RxJava的Observable进行网络请求：

ApiInterface apiInterface = RetrofitClient.getClient().create(ApiInterface.class);

apiInterface.getData()
    .subscribeOn(Schedulers.io()) // 在IO线程进行网络请求
    .observeOn(AndroidSchedulers.mainThread()) // 在主线程接收和处理结果
    .take(5) // 只接收前5个数据项
    .subscribe(dataList -> {
        // 处理获取到的数据
        for (Data data : dataList) {
            Log.d(TAG, "Received data: " + data.toString());
        }
    }, throwable -> {
        // 处理错误
        Log.e(TAG, "Error: " + throwable.getMessage());
    });

先建了一个ApiInterface的实例，用于定义网络接口。

然后用Retrofit和RxJava的Observable结合进行网络请求。通过subscribeOn()方法指定在IO线程进行网络请求，observeOn()方法指定在主线程接收和处理结果。使用take(5)操作符来限制只接收前5个数据项，即结果数量限制为5。

最后，通过subscribe方法订阅Observable，并在订阅中处理获取到的数据或错误。

concat

concat()操作符是RxJava中常用的操作符之一，用于将多个Observable按顺序连接在一起，并依次发射数据。它可以在安卓开发中用于实现多个下载任务的顺序执行。

在安卓开发中，有时需要进行多个文件的下载操作，可以使用concat操作符来依次执行下载任务：

创建一个下载任务的接口：

public interface DownloadService {
    @GET
    Observable<ResponseBody> downloadFile(@Url String fileUrl);
}

Retrofit结合RxJava的Observable进行下载任务：

DownloadService service = RetrofitClient.getClient().create(DownloadService.class);

Observable<ResponseBody> downloadTask1 = service.downloadFile("http://example.com/file1");
Observable<ResponseBody> downloadTask2 = service.downloadFile("http://example.com/file2");
Observable<ResponseBody> downloadTask3 = service.downloadFile("http://example.com/file3");

Observable.concat(downloadTask1, downloadTask2, downloadTask3)
    .subscribeOn(Schedulers.io()) // 在IO线程进行下载任务
    .observeOn(AndroidSchedulers.mainThread()) // 在主线程接收和处理结果
    .subscribe(responseBody -> {
        // 处理下载完成的文件
        saveFile(responseBody);
    }, throwable -> {
        // 处理错误
        Log.e(TAG, "Error: " + throwable.getMessage());
    });