如何在使用rxjava时取消慢发射

import io.reactivex.Observable;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public class RxJavaCancellationExample {
    private CompositeDisposable compositeDisposable = new CompositeDisposable();

    public void startSlowEmission() {
        Observable<Integer> observable = Observable.create(emitter -> {
            for (int i = 1; i <= 10; i++) {
                Thread.sleep(1000); // 模拟慢发射
                emitter.onNext(i);
            }
            emitter.onComplete();
        });

        Disposable disposable = observable
                .subscribeOn(Schedulers.io())
                .subscribe(
                        value -> System.out.println("Received: " + value),
                        Throwable::printStackTrace,
                        () -> System.out.println("Completed")
                );

        compositeDisposable.add(disposable);
    }

    public void cancelSlowEmission() {
        compositeDisposable.dispose();
    }

    public static void main(String[] args) {
        RxJavaCancellationExample example = new RxJavaCancellationExample();
        example.startSlowEmission();

        // 模拟等待一段时间后取消慢发射
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        example.cancelSlowEmission();
    }
}