使用 EnumeratorCancellation 返回 AsyncEnumerable 或循环 WithCancellation 有什么区别

Question

我有以下方法可以从 http 流中读取 csv 文档

public async IAsyncEnumerable<Line> GetLines([EnumeratorCancellation] CancellationToken cancellationToken)
{
    HttpResponseMessage response = GetResponse();

    using var responseStream = await response.Content.ReadAsStreamAsync();
    using var streamReader = new StreamReader(responseStream);
    using var csvReader = new CsvReader(streamReader);

    while (!cancellationToken.IsCancellationRequested && await csvReader.ReadAsync())
    {
        yield return csvReader.GetRecord<Line>();
    }
}

以及其他地方使用结果的方法

var documentAsyncEnumerable = graphClient.GetLines(cancellationToken);
await foreach (var document in documentAsyncEnumerable.WithCancellation(cancellationToken))
{
    // Do something with document    
}

我的问题是我应该在一个地方使用取消令牌吗？应该在产生记录之前对取消令牌采取行动还是 IAsyncEnumerable.WithCancellation() 基本上做同样的事情？如果有的话有什么区别？

Answer 1

根据消息来源GetAsyncEnumerator，在幕后，取消令牌无论如何都会传递给方法

namespace System.Collections.Generic
{
    public interface IAsyncEnumerable<out T>
    {
        IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
    }

    public interface IAsyncEnumerator<out T> : IAsyncDisposable
    {
        ValueTask<bool> MoveNextAsync();
        T Current { get; }
    }
}

你应该cancellationToken只使用一次，直接传递或使用WithCancellation，这些方法都是一样的。WithCancellation是 的扩展方法IAsyncEnumerable<T>，接受 aCancellationToken作为参数（它使用与 相同的模式ConfigureAwait）。如果[EnumeratorCancellation]编译器生成将令牌传递给GetAsyncEnumerator方法的代码

MSDN杂志中描述了两种不同方式的原因

为什么有两种不同的方法呢？将令牌直接传递给方法更容易，但是当您从其他来源获得任意 IAsyncEnumerable 但仍希望能够请求取消组成它的所有内容时，它不起作用。在极端情况下，将令牌传递给 GetAsyncEnumerator 也可能是有利的，因为这样做可以避免在单个可枚举项将被多次枚举的情况下“烧入”令牌：通过将其传递给 GetAsyncEnumerator，不同的令牌可以每次都通过。

Answer 2

我的问题是我应该只在一处使用取消令牌吗？

取消是合作性的，因此为了能够取消，您必须在提供. 所以，制作人是一处。GetLinesIAsyncEnumerable<Line>

现在，想象一下使用该数据执行某些操作的代码被命名的方法ConsumeLines，假设它是一个消费者。在您的情况下，它可能是一个代码库，但一般来说，它可能是另一个库、另一个存储库、另一个代码库。

在其他代码库中，不能保证它们具有相同的 CancellationToken.

那么，消费者如何取消呢？

消费者需要将 a 传递CancellationToken给IAsyncEnumerable<T>.GetAsyncEnumerator，但如果您使用构造，则它不会直接公开await foreach。

为了解决这个问题，WithCancellation添加了扩展方法。它只是通过将传递给它的数据包装在ConfiguredCancelableAsyncEnumerableCancellationToken中来将其转发到底层。IAsyncEnumerable

根据多种条件，它使用CreateLinkedTokenSourceCancellationToken链接到生产者中的一个，以便消费者可以使用生产者中实现的协作取消来取消，这样我们不仅可以取消消费，还可以取消生产。

是否应该在生成记录之前对取消令牌进行操作，或者 IAsyncEnumerable.WithCancellation() 基本上做同样的事情吗？如果有的话有什么区别？

是的，您应该在生产者CancellationToken代码中使用IsCancellationRequested或ThrowIfCancellationRequested来执行操作。取消是合作性的，如果你不在生产者中实现它，你将无法取消生产的值。IAsyncEnumerable

至于到底什么时候取消——在让步之前还是之后——完全取决于你，我们的想法是避免任何不必要的工作。本着这种精神，您还可以在方法的第一行检查取消情况，以避免发送不必要的 http 请求。

请记住，取消值的消耗不一定与取消值的生成相同。

同样，生产者和消费者可能位于不同的代码库中，并且可能使用CancellationTokens不同的CancellationTokenSources.

要将这些不同的链接 CancellationTokens 在一起，您需要使用 EnumeratorCancellation 属性。

请阅读我的文章 EnumeratorCancellation 中更深入的解释：生成的 IAsyncEnumerable.GetAsyncEnumerator 中的 CancellationToken 参数将不被使用