c# - Rx 和任务 - 生成新任务时取消正在运行的任务？

Question

我有一个用户交互场景，我想用 Rx 处理。

该场景类似于规范的“当用户停止输入时，做一些工作”(通常，搜索用户到目前为止输入的内容)(1) - 但我还需要:

• (2) 只获取“do some work”单元的最新结果(见下文)
• (3) 当一个新的工作单元开始时，取消任何正在进行的工作(在我的例子中它是 CPU 密集型的)

对于 (1)，我使用 IObservable对于用户事件，使用 .Throttle() 进行限制仅触发事件之间的暂停(“用户停止输入”)。

由此，我 .Select(_ => CreateMyTask(...).ToObservable()) .

这给了我一个 IObservable<IObservable<T>>其中每个内部可观察对象都包含一个任务。

为了获得 (2) 我终于申请了 .Switch()只从最新的工作单元中获取结果。

那么 (3) - 取消挂起的任务呢？

如果我没理解错的话，每当有一个新的内部IObservable<T> , .Switch()方法订阅它并取消订阅以前的一个，导致他们Dispose() .
也许可以通过某种方式连接以触发任务取消？

Answer 1

您可以只使用 Observable.FromAsync，它将生成在观察者取消订阅时取消的 token :

input.Throttle(...)
     .Select(_ => Observable.FromAsync(token => CreateMyTask(..., token)))
     .Switch()
     .Subscribe(...);

这将为每个工作单元生成一个新 token ，并在每次 Switch 切换到新 token 时取消它。