RxJava怎么使用

目录

1. RxJava简介
2. RxJava的核心概念
   • Observable
   • Observer
   • Subscriber
   • Scheduler
   • Operators
3. RxJava的基本使用
   • 创建Observable
   • 订阅Observable
   • 取消订阅
4. RxJava的操作符
   • 创建操作符
   • 变换操作符
   • 过滤操作符
   • 组合操作符
   • 错误处理操作符
   • 条件操作符
   • 数学和聚合操作符
   • 连接操作符
   • 背压操作符
5. RxJava的线程调度
   • Scheduler的类型
   • 线程切换
6. RxJava的背压
   • 背压问题
   • 背压策略
7. RxJava的实战应用
   • 网络请求
   • 数据库操作
   • UI事件处理
   • 定时任务
8. RxJava的优缺点
   • 优点
   • 缺点
9. RxJava与其他响应式编程库的比较
   • RxJava vs Reactor
   • RxJava vs Kotlin Flow
10. 总结

RxJava简介

RxJava是一个基于观察者模式的异步编程库，它扩展了观察者模式以支持数据/事件序列，并添加了操作符来对这些序列进行组合、转换和过滤。RxJava的核心思想是将异步操作和事件处理通过流式API进行表达，使得代码更加简洁、易读和易于维护。

RxJava最初由Netflix开发，并在2013年开源。它基于ReactiveX（Reactive Extensions）项目，该项目最初由微软开发，旨在为.NET平台提供响应式编程支持。RxJava是ReactiveX在Java平台上的实现，后来也被移植到了其他语言和平台，如RxJS（JavaScript）、RxSwift（Swift）等。

RxJava的核心概念

Observable

Observable是RxJava中的核心类，它代表一个可观察的数据源。Observable可以发射零个或多个数据项，然后完成或出错。Observable可以被订阅，订阅者（Observer或Subscriber）会接收到Observable发射的数据项。

Observer

Observer是Observable的订阅者，它定义了如何处理Observable发射的数据项。Observer接口包含四个方法：

• onNext(T item): 当Observable发射一个数据项时调用。
• onError(Throwable error): 当Observable发生错误时调用。
• onComplete(): 当Observable完成发射数据项时调用。
• onSubscribe(Disposable d): 当Observer订阅Observable时调用。

Subscriber

Subscriber是Observer的一个实现类，它提供了更多的控制能力，例如取消订阅。Subscriber通常用于需要手动管理订阅的场景。

Scheduler

Scheduler是RxJava中用于控制线程调度的类。它允许开发者在不同的线程上执行Observable的操作，例如在后台线程执行耗时操作，在主线程更新UI。

Operators

操作符是RxJava中用于对Observable发射的数据流进行变换、过滤、组合等操作的方法。RxJava提供了丰富的操作符，使得开发者可以轻松地处理复杂的数据流。

RxJava的基本使用

创建Observable

在RxJava中，Observable可以通过多种方式创建。以下是几种常见的创建方式：

1. Observable.create(): 通过手动调用Emitter的方法来发射数据项。

   Observable<Integer> observable = Observable.create(emitter -> {
       emitter.onNext(1);
       emitter.onNext(2);
       emitter.onNext(3);
       emitter.onComplete();
   });

1. Observable.just(): 创建一个发射指定数据项的Observable。

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

1. Observable.fromIterable(): 从Iterable对象创建Observable。

   List<Integer> list = Arrays.asList(1, 2, 3);
   Observable<Integer> observable = Observable.fromIterable(list);

1. Observable.range(): 创建一个发射指定范围内的整数序列的Observable。

   Observable<Integer> observable = Observable.range(1, 3);

1. Observable.interval(): 创建一个按固定时间间隔发射递增整数的Observable。

   Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS);

订阅Observable

创建Observable后，可以通过subscribe()方法订阅它。subscribe()方法有多个重载版本，可以接受Observer、Subscriber或Lambda表达式作为参数。

observable.subscribe(
    item -> System.out.println("onNext: " + item),
    error -> System.err.println("onError: " + error),
    () -> System.out.println("onComplete")
);

取消订阅

在某些情况下，可能需要手动取消订阅以避免内存泄漏。可以通过Disposable对象来取消订阅。

Disposable disposable = observable.subscribe(
    item -> System.out.println("onNext: " + item),
    error -> System.err.println("onError: " + error),
    () -> System.out.println("onComplete")
);

// 取消订阅
disposable.dispose();

RxJava的操作符

RxJava提供了丰富的操作符，用于对Observable发射的数据流进行各种操作。以下是一些常见的操作符分类及其示例。

创建操作符

1. create(): 手动创建Observable。

   Observable<Integer> observable = Observable.create(emitter -> {
       emitter.onNext(1);
       emitter.onNext(2);
       emitter.onNext(3);
       emitter.onComplete();
   });

1. just(): 创建一个发射指定数据项的Observable。

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

1. fromIterable(): 从Iterable对象创建Observable。

   List<Integer> list = Arrays.asList(1, 2, 3);
   Observable<Integer> observable = Observable.fromIterable(list);

1. interval(): 创建一个按固定时间间隔发射递增整数的Observable。

   Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS);

1. range(): 创建一个发射指定范围内的整数序列的Observable。

   Observable<Integer> observable = Observable.range(1, 3);

变换操作符

1. map(): 对Observable发射的每个数据项应用一个函数，并将结果发射出去。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .map(item -> item * 2);

1. flatMap(): 将Observable发射的每个数据项转换为另一个Observable，然后将这些Observable发射的数据项合并。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .flatMap(item -> Observable.just(item * 2));

1. concatMap(): 类似于flatMap，但保持原始顺序。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .concatMap(item -> Observable.just(item * 2));

1. switchMap(): 类似于flatMap，但只保留最新的Observable发射的数据项。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .switchMap(item -> Observable.just(item * 2));

1. buffer(): 将Observable发射的数据项按指定大小或时间间隔进行缓冲，并将缓冲后的数据项列表发射出去。

   Observable<List<Integer>> observable = Observable.range(1, 10)
       .buffer(3);

过滤操作符

1. filter(): 过滤掉不符合条件的数据项。

   Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5)
       .filter(item -> item % 2 == 0);

1. take(): 只发射前N个数据项。

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

1. skip(): 跳过前N个数据项。

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

1. distinct(): 过滤掉重复的数据项。

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

1. debounce(): 只在指定的时间间隔内没有发射新数据项时，才发射最后一个数据项。

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

组合操作符

1. merge(): 将多个Observable合并为一个Observable。

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

1. concat(): 将多个Observable按顺序连接起来。

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

1. zip(): 将多个Observable发射的数据项按顺序组合成一个新的数据项。

   Observable<Integer> observable1 = Observable.just(1, 2, 3);
   Observable<String> observable2 = Observable.just("A", "B", "C");
   Observable<String> observable = Observable.zip(observable1, observable2,
       (item1, item2) -> item1 + item2);

1. combineLatest(): 当任何一个Observable发射新数据项时，将最新的数据项组合成一个新的数据项。

   Observable<Integer> observable1 = Observable.just(1, 2, 3);
   Observable<String> observable2 = Observable.just("A", "B", "C");
   Observable<String> observable = Observable.combineLatest(observable1, observable2,
       (item1, item2) -> item1 + item2);

1. switchOnNext(): 将多个Observable合并为一个Observable，并只保留最新的Observable发射的数据项。

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

错误处理操作符

1. onErrorReturn(): 当发生错误时，返回一个默认值。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .map(item -> {
           if (item == 2) throw new RuntimeException("Error");
           return item;
       })
       .onErrorReturn(error -> -1);

1. onErrorResumeNext(): 当发生错误时，切换到另一个Observable。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .map(item -> {
           if (item == 2) throw new RuntimeException("Error");
           return item;
       })
       .onErrorResumeNext(Observable.just(4, 5, 6));

1. retry(): 当发生错误时，重试指定次数。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .map(item -> {
           if (item == 2) throw new RuntimeException("Error");
           return item;
       })
       .retry(2);

1. retryWhen(): 当发生错误时，根据指定的条件重试。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .map(item -> {
           if (item == 2) throw new RuntimeException("Error");
           return item;
       })
       .retryWhen(errors -> errors.zipWith(Observable.range(1, 3), (error, retryCount) -> retryCount));

条件操作符

1. all(): 判断Observable发射的所有数据项是否都满足指定条件。

   Observable<Boolean> observable = Observable.just(1, 2, 3)
       .all(item -> item > 0);

1. any(): 判断Observable发射的数据项中是否有任何一个满足指定条件。

   Observable<Boolean> observable = Observable.just(1, 2, 3)
       .any(item -> item > 2);

1. contains(): 判断Observable发射的数据项中是否包含指定值。

   Observable<Boolean> observable = Observable.just(1, 2, 3)
       .contains(2);

1. isEmpty(): 判断Observable是否没有发射任何数据项。

   Observable<Boolean> observable = Observable.just(1, 2, 3)
       .isEmpty();

1. defaultIfEmpty(): 如果Observable没有发射任何数据项，则发射一个默认值。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .filter(item -> item > 3)
       .defaultIfEmpty(-1);

数学和聚合操作符

1. count(): 统计Observable发射的数据项数量。

   Observable<Long> observable = Observable.just(1, 2, 3)
       .count();

1. reduce(): 对Observable发射的数据项进行累积操作。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .reduce((acc, item) -> acc + item);

1. scan(): 对Observable发射的数据项进行累积操作，并发射每次累积的结果。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .scan((acc, item) -> acc + item);

1. sum(): 对Observable发射的数据项求和。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .sumInt(item -> item);

1. average(): 对Observable发射的数据项求平均值。

   Observable<Double> observable = Observable.just(1, 2, 3)
       .averageDouble(item -> item);

连接操作符

1. publish(): 将Observable转换为ConnectableObservable，允许多个订阅者共享同一个Observable。

   ConnectableObservable<Integer> connectableObservable = Observable.just(1, 2, 3)
       .publish();
   connectableObservable.connect();

1. replay(): 将Observable转换为ConnectableObservable，并在订阅时重放之前发射的数据项。

   ConnectableObservable<Integer> connectableObservable = Observable.just(1, 2, 3)
       .replay();
   connectableObservable.connect();

1. refCount(): 将ConnectableObservable转换为普通的Observable，并在有订阅者时自动连接。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .publish()
       .refCount();

1. autoConnect(): 将ConnectableObservable转换为普通的Observable，并在指定数量的订阅者订阅时自动连接。

   Observable<Integer> observable = Observable.just(1, 2, 3)
       .publish()
       .autoConnect(2);

背压操作符

1. onBackpressureBuffer(): 当Observable发射数据项的速度超过订阅者处理的速度时，将数据项缓冲起来。

   Observable<Integer> observable = Observable.range(1, 1000)
       .onBackpressureBuffer();

1. onBackpressureDrop(): 当`Observable