java - 如何中断已发布的 Flowable

Question

如何中断长期任务？

public Flowable<Integer> simple(int number) {
    Flowable<Integer> flowable = Flowable.create(emitter -> {
        for (int i = 1; i <= number; ++i) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw e;
            }
            emitter.onNext(i);
        }
        emitter.onComplete();
    }, BackpressureStrategy.BUFFER);
    flowable = flowable.subscribeOn(Schedulers.newThread());
    return flowable.publish().autoConnect();
}

在发布 Flowable 后调用 dispose 方法后，任务将继续。所以我想捕获一个被中断的异常。

Flowable<Integer> f = simple(100);
Disposable d = f.subscribe(number -> System.out.println(number));
Thread.sleep(500);
d.dispose();

它也具有与重播相同的行为。

Answer 1

存储您正在执行的线程并注册一次性事件以中断线程。

Flowable<Integer> flowable = Flowable.create(emitter -> {
    Thread myThread = Thread.currentThread();
    emitter.setCancellable(() -> myThread.interrupt());

    for (int i = 1; i <= number; ++i) {
        if (Thread.interrupted()) {
            // Handle interrupt happening while Thread.sleep
            //  was not executing here.
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw e;
        }
        emitter.onNext(i);
    }
    emitter.onComplete();
}, BackpressureStrategy.BUFFER);