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