c# - 可观察的订阅如何优雅地终止？

Question

我正在尝试使用 Reactive Extensions (Rx) 来处理数据流。但是，每个元素的处理可能需要一些时间。为了中断处理，我使用了一个 CancellationToken，它有效地停止了订阅。

当请求取消时，如何在不丢失任何数据的情况下优雅地完成当前工作并正确终止？

例子

var cts = new CancellationTokenSource();
cts.Token.Register(() => Console.WriteLine("Token cancelled."));

var observable = Observable
    .Interval(TimeSpan.FromMilliseconds(250));

observable
    .Subscribe(
        value =>
            {
                Console.WriteLine(value);
                Thread.Sleep(500); // Simulate processing
                
                if (cts.Token.IsCancellationRequested)
                {
                    Console.WriteLine("Cancellation detected on {0}.", value);
                    Thread.Sleep(500); // Simulate some time consuming shutdown
                    Console.WriteLine("Cleaning up done for {0}.", value);
                }
            },
        () => Console.WriteLine("Completed"),
        cts.Token);
        
Console.ReadLine();
cts.Cancel();
Console.WriteLine("Job terminated.");

输出

0
1
2
Token cancelled.
Job terminated.
Cancellation detected on 2.
Cleaning up done for 2.

从输出中可以看出，“Job terminated”这行不是最后一行，这意味着清理在应用程序终止之前没有足够的时间完成。

预期输出

0
1
2
Token cancelled.
Cancellation detected on 2.
Cleaning up done for 2.
Job terminated.

“作业终止”行是要打印的最后一行。 “取消”和“清洁”行已被允许慢慢来。

(编辑:添加预期输出)

Answer 1

如果我对问题的理解正确，这不是 Rx 问题，而是“无论你在订阅中做什么”问题。您的订阅操作需要半秒，清理可能需要另外半秒，而您的作业终止需要几微秒。您希望在取消和终止之间挤出什么？

我能给你的最好建议是让订阅操作比 Thread.Sleep 调用更好地尊重取消 token 。