马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。
您需要 登录 才可以下载或查看，没有账号？立即注册
×
在Android开发中，RxJava是一个盛行的库，用于处理惩罚异步和基于变乱的编程。RxJava 中的 Emitter 是一个接口(组件)，用于发射 Observable 的变乱（onNext、onComplete、onError）。它是创建 Observable 时的核心组件，允许你手动控制数据流。根据差别场景，RxJava 提供了三种 Emitter：
- SyncEmitter：同步发射变乱，必须依照 RxJava 的规则（如在 onComplete/onError 后不再发射变乱）。
- AsyncEmitter：异步发射变乱，支持背压（Backpressure）。
- SerializedEmitter：线程安全的发射器，主动处理惩罚多线程并发题目。
Emitter有几个方法，可以发射差别范例的变乱：
- onNext(Object): 发射一个非竣事变乱。
- onCompleted(): 发射一个完成变乱，体现变乱流竣事。
- onError(Throwable): 发射一个错误变乱，体现变乱流出现非常。
根本用法示例
下面是一个利用 Observable.create() 和 Emitter 创建自界说 Observable 的示例：- import io.reactivex.rxjava3.core.Observable;
- import io.reactivex.rxjava3.core.Observer;
- import io.reactivex.rxjava3.disposables.Disposable;
- import io.reactivex.rxjava3.subjects.PublishSubject;
- public class EmitterExample {
- public static void main(String[] args) {
- // 创建一个 Observable，使用 Emitter 发射数据
- Observable<Integer> source = Observable.create(emitter -> {
- try {
- // 发射数据项
- emitter.onNext(1);
- emitter.onNext(2);
- emitter.onNext(3);
-
- // 检查订阅状态，避免在取消订阅后继续发射
- if (!emitter.isDisposed()) {
- emitter.onComplete();
- }
- } catch (Exception e) {
- // 发生错误时发射错误
- emitter.onError(e);
- }
- });
- // 创建一个 Observer 来处理发射的事件
- Observer<Integer> observer = new Observer<Integer>() {
- @Override
- public void onSubscribe(Disposable d) {
- System.out.println("onSubscribe");
- }
- @Override
- public void onNext(Integer value) {
- System.out.println("onNext: " + value);
- }
- @Override
- public void onError(Throwable e) {
- System.err.println("onError: " + e.getMessage());
- }
- @Override
- public void onComplete() {
- System.out.println("onComplete");
- }
- };
- // 订阅 Observable
- source.subscribe(observer);
- }
- }
- 服从发射规则：
- 必须在 onComplete 或 onError 后制止发射数据。
- 不能同时调用 onComplete 和 onError。
- 一旦调用 onComplete 或 onError，Observable 的生命周期竣事。
- 处理惩罚背压：
如果异步发射数据且鄙俚处理惩罚速率慢，需利用支持背压的操纵符（如 Flowable）或得当的战略（如 buffer、sample）。
- 资源管理：
在 onSubscribe 中返回的 Disposable 可用于取消订阅，防止内存走漏。比方：
- emitter.setCancellable(() -> {
- // 释放资源的代码
- });
- 线程安全：
如果在多线程情况中利用 Emitter，发起利用 Observable.create() 共同 serialize() 操纵符确保线程安全。
Emitter 是 RxJava 中强大的工具，但需审慎利用以制止违反 RxJava 的规则导致难以调试的题目。
RxJava 的核心概念远不止 Emitter，它基于相应式编程范式，通过异步数据流将生产者和斲丧者解耦。以下是 RxJava 中其他关键的核心概念：
1. Observable / Flowable
脚色：数据的生产者，负责发失变乱序列。
区别：
- Observable：不支持背压（Backpressure），实用于变乱数量较少或处理惩罚速率快的场景。
- Flowable：支持背压，通过 Subscriber 控制数据流速，防止数据过载。
示例：- // 创建 Observable（不支持背压）
- Observable.just("A", "B", "C")
- .subscribe(System.out::println);
- // 创建 Flowable（支持背压）
- Flowable.range(1, 1000)
- .subscribe(
- System.out::println, // onNext
- Throwable::printStackTrace, // onError
- () -> System.out.println("Completed") // onComplete
- );
脚色：数据的斲丧者，监听 Observable/Flowable 发出的变乱。
核心方法：
- onNext(T value)：吸收数据项。
- onError(Throwable e)：处理惩罚错误并制止流。
- onComplete()：流正常竣事时调用。
示例：- Observer<String> observer = new Observer<String>() {
- @Override
- public void onSubscribe(Disposable d) {
- System.out.println("订阅成功");
- }
- @Override
- public void onNext(String value) {
- System.out.println("接收到: " + value);
- }
- @Override
- public void onError(Throwable e) {
- e.printStackTrace();
- }
- @Override
- public void onComplete() {
- System.out.println("流已完成");
- }
- };
- Observable.just("Hello", "World").subscribe(observer);
作用：管理订阅关系，允许在必要时取消订阅，防止内存走漏。
方法：
- dispose()：取消订阅。
- isDisposed()：查抄是否已取消。
示例：- Disposable disposable = Observable.interval(1, TimeUnit.SECONDS)
- .subscribe(System.out::println);
- // 5秒后取消订阅
- Thread.sleep(5000);
- disposable.dispose();
作用：控制数据流的实行线程（如 IO、盘算、UI 线程）。
常用 Scheduler：
- Schedulers.io()：实用于 IO 麋集型操纵（如网络哀求）。
- Schedulers.computation()：实用于盘算麋集型操纵。
- AndroidSchedulers.mainThread()（Android 平台）：用于 UI 更新。
示例：- Observable.just("A", "B", "C")
- .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
- .observeOn(AndroidSchedulers.mainThread()) // 指定回调在主线程
- .subscribe(System.out::println);
作用：转换、过滤、组合数据流，实现复杂的数据处理惩罚逻辑。
分类：
- 创建操纵符：just()、from()、interval()、create()。
- 转换操纵符：map()、flatMap()、buffer()、scan()。
- 过滤操纵符：filter()、take()、distinct()、debounce()。
- 组合操纵符：merge()、concat()、zip()、combineLatest()。
- 错误处理惩罚：onErrorReturn()、retry()、catchError()。
示例：- Observable.just(1, 2, 3, 4)
- .map(num -> num * 10) // 转换：每个元素乘以10
- .filter(num -> num > 20) // 过滤：只保留大于20的元素
- .subscribe(System.out::println); // 输出：30, 40
特性：既是 Observable 又是 Observer，可同时发射和吸收变乱。
常见范例：
- PublishSubject：只发射订阅后产生的变乱。
- BehaviorSubject：发射订阅前的末了一个变乱 + 订阅后的全部变乱。
- ReplaySubject：缓存全部变乱，订阅者会收到全部汗青数据。
- AsyncSubject：只发射末了一个变乱（在 onComplete() 之后）。
示例：- PublishSubject<String> subject = PublishSubject.create();
- subject.onNext("Event 1"); // 订阅前的事件不会被收到
- subject.subscribe(System.out::println); // 订阅者开始监听
- subject.onNext("Event 2"); // 输出：Event 2
- subject.onNext("Event 3"); // 输出：Event 3
题目：当 Observable 发射速率 > Observer 处理惩罚速率时，大概导致内存溢出。
办理方案：
- 利用 Flowable 更换 Observable。
- 应用背压战略：
- BUFFER：缓存全部变乱（大概 OOM）。
- DROP：扬弃无法处理惩罚的变乱。
- LATEST：只保存最新的变乱。
示例：- Flowable.range(1, 1000)
- .onBackpressureDrop() // 丢弃无法处理的事件
- .observeOn(Schedulers.io())
- .subscribe(System.out::println);
RxJava 通过这些核心概念构建了强大的异步数据流处理惩罚本领，让代码更简便、可维护，并有效管理并发和资源。把握这些概念是利用 RxJava 的关键。
|
发表于 2026-1-21 06:33:10