Rx 库是否省略了它创建的 CancellationTokenSources 的处理？

问题描述

Rx 库包含接受 lambda 参数的运算符，其中一些 lambda 表达式带有 CancellationToken，该 FromAsync 由库本身控制。这些运算符的一些示例是 CancellationTokenSource、StartAsync 和 Create：

// Converts an asynchronous action into an observable sequence. Each subscription
// to the resulting sequence causes the action to be started. The CancellationToken
// passed to the asynchronous action is tied to the observable sequence's subscription
// that triggered the action's invocation and can be used for best-effort cancellation.
public static IObservable<Unit> FromAsync(Func<CancellationToken,Task> actionAsync);

我的印象是，Rx 库在管理显然必须在幕后创建的 insists strongly 的生命周期方面做得很好，但我不再那么确定了。让我们首先声明应该处理 CancellationTokenSource 的文档 here：

这种类型实现了 Idisposable 接口。当您使用完该类型的实例后，您应该直接或间接地处理它。要直接处理该类型，请在 dispose/try 块中调用其 catch 方法。要间接处理它，请使用诸如 using（在 C# 中）或 Using（在 Visual Basic 中）之类的语言结构。

也来自_source：

在释放对 dispose 的最后一次引用之前，始终调用 CancellationTokenSource。否则，在垃圾收集器调用 CancellationTokenSource 对象的 Finalize 方法之前，它使用的资源不会被释放。

我做了下面的实验来检验我的假设。它使用反射分别读取 CancellationToken 和 CancellationTokenSource 类型的私有字段 _disposed 和 Try it on Fiddle (.NET 5)。

CancellationToken capturedToken = default;
var subscription = Observable.FromAsync(async token =>
{
    capturedToken = token;
    token.Register(() => Console.WriteLine("Token canceled"));
    await Task.Delay(Timeout.Infinite,token);
})
.takeuntil(Observable.Timer(TimeSpan.FromMilliseconds(500)))
.Finally(() => Console.WriteLine("The observable was terminated"))
.Subscribe();

Thread.Sleep(1000);

var cts = (CancellationTokenSource)(typeof(CancellationToken)
    .GetField("_source",BindingFlags.NonPublic | BindingFlags.Instance)
    .GetValue(capturedToken));
bool disposed = (bool)(typeof(CancellationTokenSource)
    .GetField("_disposed",BindingFlags.NonPublic | BindingFlags.Instance)
    .GetValue(cts));
Console.WriteLine($"IsCancellationRequested: {cts.IsCancellationRequested}");
Console.WriteLine($"Isdisposed: {disposed}");

输出：

Token canceled
The observable was terminated
IsCancellationRequested: True
Isdisposed: False

{{3}}（.NET Framework 版本，具有不同命名的私有字段）

在异步操作取消且可观察对象终止后半秒检查捕获的 CancellationToken。 _disposed 字段的值为 false，表示尚未调用关联的 dispose 的 CancellationTokenSource 方法。我做错了什么，还是 Rx 库确实忽略了处理它创建的 CancellationTokenSource ？

.NET 5.0.1、System.Reactive 5.0.0、C# 9

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）