客户端断开时如何取消可无限重复的Flowable？

你的问题出在流的生命周期设计上：原代码每次发射一个test就调用onComplete结束当前流，再通过repeat()重新订阅新的流。当客户端取消请求时，取消信号是终止整个repeat()的循环，但当前正在运行的那个create流已经正常完成了（不是被取消的），所以doOnCancel自然不会触发。

要让取消时触发doOnCancel，你需要调整流的实现逻辑：

• 不要用repeat() + onComplete的方式循环，而是在create内部持续发射数据，直到流被取消。
• 必须给发射器设置取消回调，确保取消信号能传递到上游，触发对应的逻辑。

修改后的代码如下：

Flowable.create<String>({ emitter ->
    // 设置取消回调，流被取消时会执行这里
    emitter.setCancellable {
        println("emitter canceled")
    }
    // 持续发射数据，直到流被取消
    val thread = Thread {
        while (!emitter.isCancelled) {
            emitter.onNext("test")
            Thread.sleep(1000) // 控制发射频率，避免过快
        }
    }
    thread.start()
}, BackpressureStrategy.MISSING)
.doOnCancel {
    println("canceled")
}

关键改动说明： #

1. 去掉repeat()和onComplete：现在整个流是一个持续运行的实例，不会频繁创建新流，取消信号能直接作用于这个流。
2. 添加emitter.setCancellable：明确告诉发射器，当流被取消时要执行的逻辑，同时这也是让doOnCancel能被触发的前提——只有当流真正被取消（而非正常完成）时，doOnCancel才会执行。
3. 循环发射直到取消：用emitter.isCancelled判断是否继续发射，确保客户端断开时能停止发射并触发取消回调。

这样调整后，当你在POSTMAN中取消请求时，doOnCancel里的println("canceled")就会正常执行了。

内容的提问来源于stack exchange，提问作者Ishaan Sinha