rxjava怎么取消订阅

在 RxJava 中，“取消订阅”本质上是切断上游和下游的联系，让上游不再往下发送数据，下游也不再接收。下面按常见场景给你梳理一下。

一、最基础方式：Disposable

1️⃣ subscribe() 返回 Disposable

Disposable disposable = Observable.just(1, 2, 3)
        .subscribe(
            item -> Log.d("RX", "onNext: " + item),
            Throwable::printStackTrace,
            () -> Log.d("RX", "onComplete")
        );

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

✅ dispose() 之后：

• 上游停止发射
• 下游不再收到事件

二、批量管理：CompositeDisposable

适合 Activity / Fragment / ViewModel

CompositeDisposable compositeDisposable = new CompositeDisposable();

Disposable d1 = Observable.interval(1, TimeUnit.SECONDS)
        .subscribe(aLong -> Log.d("RX", "tick"));

compositeDisposable.add(d1);

// 页面销毁时
compositeDisposable.clear();   // 保留容器
// 或
compositeDisposable.dispose(); // 容器也废了

✅ 推荐用法（Android）：

@Override
protected void onDestroy() {
    compositeDisposable.clear();
    super.onDestroy();
}

三、自动取消：绑定生命周期（推荐）

1️⃣ RxAndroid + bindToLifecycle

Observable.interval(1, TimeUnit.SECONDS)
        .compose(bindToLifecycle(this))
        .subscribe(...);

2️⃣ RxLifecycle

Observable.timer(3, TimeUnit.SECONDS)
        .compose(RxLifecycle.bindUntilEvent(lifecycle, ActivityEvent.DESTROY))
        .subscribe(...);

四、条件取消：takeUntil

Observable.interval(1, TimeUnit.SECONDS)
        .takeUntil(Observable.timer(5, TimeUnit.SECONDS))
        .subscribe(...);

✅ 5 秒后自动取消订阅

五、在 Observer 内部取消

Observable.range(1, 10)
        .subscribe(new Observer<Integer>() {
            private Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                if (integer == 5) {
                    disposable.dispose();
                }
            }

            @Override public void onError(Throwable e) {}
            @Override public void onComplete() {}
        });

六、Flowable（背压）也一样

Flowable.range(1, 100)
        .subscribe(new Subscriber<Integer>() {
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription s) {
                subscription = s;
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Integer integer) {
                if (integer == 3) {
                    subscription.cancel();
                }
            }
        });

总结

如果你是用 RxJava2 / RxJava3 / Android / Kotlin，可以告诉我，我可以给你更具体的写法。