如何中断发布的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);

如何中断发布的flowable

How to interrupt the published flowable

java

rx-java

rx-android