c# - 在返回带取消的 IAsyncEnumerable 的函数中迭代 IAsyncEnumerable

Question

正如标题所说，我必须执行以下功能:

public async IAsyncEnumerable<Job> GetByPipeline(int pipelineId,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    await foreach (var job in context.Jobs.Where(job => job.Pipeline.Id == pipelineId)
        .AsAsyncEnumerable()
        .WithCancellation(cancellationToken)
        .ConfigureAwait(false))
    {
        yield return job;
    }
}

我无法理解取消 token 的去向，并且有一种挥之不去的感觉，即我在太多地方使用了它。

当您解构所有花哨的异步内容时，这里实际上发生了什么？有没有更好的方法来编写这个函数？

Answer 1

对于初学者来说，这个方法可以简化为:

public IAsyncEnumerable<Job> GetByPipeline(int pipelineId)
{
    return context.Jobs
                  .Where(job => job.Pipeline.Id == pipelineId)
                  .AsAsyncEnumerable();
}

甚至

public IAsyncEnumerable<Job> GetByPipeline(int pipelineId)
    => context.Jobs
              .Where(job => job.Pipeline.Id == pipelineId)
              .AsAsyncEnumerable();

该方法不对 job 执行任何操作，因此不需要对其进行迭代。

取消

如果该方法实际使用了job，应该在哪里使用取消标记？

让我们稍微清理一下这个方法。等价于:

public async IAsyncEnumerable<Job> GetByPipeline(
      int pipelineId, 
      [EnumeratorCancellation] CancellationToken ct = default)
{
    //Just a query, doesn't execute anything
    var query =context.Jobs.Where(job => job.Pipeline.Id == pipelineId);

    //Executes the query and returns the *results* as soon as they arrive in an async stream
    var jobStream=query.AsAsyncEnumerable();

    //Process the results from the async stream as they arrive
    await foreach (var job in jobStream.WithCancellation(ct).ConfigureAwait(false))
    {
        //Does *that* need cancelling?
        DoSometingExpensive(job);
    }
}

IQueryable query 不运行任何东西，它表示查询。它不需要取消。

AsAsyncEnumerable()、AsEnumerable()、ToList() 等执行查询并返回一些结果. ToList() 等使用所有结果，而 As...Enumerable() 方法仅在请求时产生结果。查询无法取消，As_Enumerable() 方法不会返回任何内容，除非被要求，因此它们不需要取消。

await foreach 将遍历整个异步流，因此如果我们希望能够中止它，我们确实需要传递取消 token 。

最后，DoSometingExpensive(job); 是否需要取消？是不是太贵了，如果时间太长，我们希望能够摆脱它？或者我们可以等到它完成后再退出循环吗？如果需要取消，它也需要 CancellationToken。

配置等待

最后，ConfigureAwait(false) 不参与取消，可能根本不需要。没有它，在每次 await 执行后返回到原始同步上下文。在桌面应用程序中，这意味着 UI 线程。这就是允许我们在异步事件处理程序中修改 UI 的原因。

如果 GetByPipeline 在桌面应用程序上运行并想要修改 UI，则必须删除 ConfugureAwait :

await foreach (var job in jobStream.WithCancellation(ct))
{
        //Update the UI
        toolStripProgressBar.Increment(1);
        toolStripStatusLabel.Text=job.Name;
        //Do the actual job
        DoSometingExpensive(job);
}

使用 ConfigureAwait(false)，在线程池线程上继续执行，我们不能触摸 UI。

库代码不应影响执行的恢复方式，因此大多数库使用 ConfigureAwait(false) 并将最终决定留给 UI 开发人员。

如果 GetByPipeline 是库方法，请使用 ConfigureAwait(false)。