Observable.Create：CancellationToken 不会转换为 IsCancellationRequested

Question

拿这个小脚本（在 LINQPad 中设计，但应该在任何地方运行）：

void Main()
{
    Task.Run(() => Worker()).Wait();
}

async Task Worker()
{
    if (SynchronizationContext.Current != null)
        throw new InvalidOperationException("Don't want any synchronization!");

    BaseClass provider = new Implementation();
    Func<IObserver<TimeSpan>, CancellationToken, Task> subscribeAsync =
        provider.CreateValues;
    var observable = Observable.Create(subscribeAsync);

    var cancellation = new CancellationTokenSource(5500).Token; // gets cancelled after 5.5s
    cancellation.Register(() => Console.WriteLine("token is cancelled now"));
    await observable
        .Do(ts =>
        {
            Console.WriteLine("Elapsed: {0}; cancelled: {1}",
                ts,
                cancellation.IsCancellationRequested);
            cancellation.ThrowIfCancellationRequested();
        })
        .ToTask(cancellation)
        .ConfigureAwait(false);
}

abstract class BaseClass
{
    // allow implementers to use async-await
    public abstract Task CreateValues(IObserver<TimeSpan> observer, CancellationToken cancellation);
}

class Implementation : BaseClass
{
    // creates Values for 10s; entirely CPU-bound: no way for async-await hence return Task.CompletedTask
    public override Task CreateValues(IObserver<TimeSpan> observer, CancellationToken cancellation)
    {
        try
        {
            var sw = Stopwatch.StartNew();
            for (int i = 0; i < 10; i++)
            {
                for (int j = 0; j < 3; j++)
                {
                    Console.WriteLine("{0}/{1} cancelled:{2}", i, j, cancellation.IsCancellationRequested);
                    Thread.Sleep(333);
                }

                if (cancellation.IsCancellationRequested) // !! never gets true !!
                    throw new ApplicationException("token is cancelled");

                observer.OnNext(sw.Elapsed);
            }

            return Task.CompletedTask;
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex);
            throw;
        }
    }
}

该方法Implementation.CreateValues只是在整个 10 秒内保持运行，而不是在 5.5 秒后停止。将CancellationToken在过去了Observable.Create，甚至不过渡到取消状态（当然原来的令牌一样）！

这是一个错误吗？做错事是我的错吗？

输出是：

0/0 cancelled:False
0/1 cancelled:False
0/2 cancelled:False
Elapsed: 00:00:01.0205951; cancelled: False
1/0 cancelled:False
1/1 cancelled:False
1/2 cancelled:False
Elapsed: 00:00:02.0253279; cancelled: False
2/0 cancelled:False
2/1 cancelled:False
2/2 cancelled:False
Elapsed: 00:00:03.0274035; cancelled: False
3/0 cancelled:False
3/1 cancelled:False
3/2 cancelled:False
Elapsed: 00:00:04.0294796; cancelled: False
4/0 cancelled:False
4/1 cancelled:False
4/2 cancelled:False
Elapsed: 00:00:05.0315332; cancelled: False
5/0 cancelled:False
5/1 cancelled:False
token is cancelled now
5/2 cancelled:False
Elapsed: 00:00:06.0335601; cancelled: True
6/0 cancelled:False
6/1 cancelled:False
6/2 cancelled:False
Elapsed: 00:00:07.0436211; cancelled: True
7/0 cancelled:False
7/1 cancelled:False
7/2 cancelled:False
Elapsed: 00:00:08.0457921; cancelled: True
8/0 cancelled:False
8/1 cancelled:False
8/2 cancelled:False
Elapsed: 00:00:09.0477509; cancelled: True
9/0 cancelled:False
9/1 cancelled:False
9/2 cancelled:False
Elapsed: 00:00:10.0498751; cancelled: True
[AggregateException] at Main/Task.Wait()

Answer 1

传递给subscribeAsync函数的取消令牌由Observable.Create调用实例化，而不是您正在实例化的取消令牌。

根据Observable.Create过载摘要：

从指定的可取消异步订阅方法创建可观察序列。传递给异步 Subscribe 方法的 CancellationToken 与返回的一次性订阅相关联，允许尽力取消。

简而言之，取消令牌将在您处理订阅时取消，而不是在指定的延迟之后。

您应该能够如下重构您的代码以使其工作：

Observable.Create(observer => subscribeAsync(observer, cancellation));

希望能帮助到你。