C# 异步迭代器为什么需要 EnumeratorCancellation？

前言

为什么需要 EnumeratorCancellation？

在使用 C# 编写异步迭代器时，您可能会遇到如下警告：

warning CS8425: 异步迭代器“TestConversationService.ChatStreamed(IReadOnlyList<ChatMessage>, ChatCompletionOptions, CancellationToken)”具有一个或多个类型为 “CancellationToken” 的参数，但它们都未用 “EnumeratorCancellation” 属性修饰，因此将不使用所生成的 “IAsyncEnumerable<>.GetAsyncEnumerator” 中的取消令牌参数。

看到这样的警告，您可能会困惑：究竟需要在异步迭代器的方法参数上添加 [EnumeratorCancellation] 属性吗？如果不添加，会有什么区别？ 让我们深入探讨一下这个问题，揭示其背后的真相。

正常调用时，[EnumeratorCancellation] 的影响

如果您只是简单地在异步迭代器方法中传递一个普通的 CancellationToken，无论是否使用 [EnumeratorCancellation]，方法的行为似乎并没有显著区别。例如：

public async IAsyncEnumerable<int> GenerateNumbersAsync(CancellationToken cancellationToken = default)
{
for (int i = 0; i < 10; i++)
{
cancellationToken.ThrowIfCancellationRequested();
yield return i;
await Task.Delay(1000, cancellationToken);
}
}

public async Task ConsumeNumbersAsync()
{
CancellationTokenSource cts = new CancellationTokenSource();
Task cancelTask = Task.Run(async () =>
{
await Task.Delay(3000);
cts.Cancel();
});

try
{
await foreach (var number in GenerateNumbersAsync(cts.Token))
{
Console.WriteLine(number);
}
}
catch (OperationCanceledException)
{
Console.WriteLine(“枚举已被取消”);
}

await cancelTask;
}

输出如下：

0
1
2
枚举已被取消

在上述代码中，即使没有使用 [EnumeratorCancellation]，取消令牌 cts.Token 依然会生效，导致迭代过程被取消。这可能会让开发者误以为 [EnumeratorCancellation] 没有实际作用，进而引发更多的困惑。

揭开真相：生产者与消费者的职责分离

实际上，[EnumeratorCancellation] 的核心作用在于 实现生产者与消费者的职责分离。具体来说：

生产者（即提供数据的异步迭代方法）专注于数据的生成和响应取消请求，不关心取消请求的来源或何时取消。

消费者（即使用数据的部分）负责控制取消逻辑，独立地决定何时取消整个迭代过程。

通过这种设计，生产者不需要知道取消请求是由谁或何时发起的，简化了生产者的设计，同时赋予消费者更大的控制权。这不仅提高了代码的可维护性和可复用性，还避免了取消逻辑的混乱。

示例说明

下面通过一个示例，直观地展示 [EnumeratorCancellation] 如何实现职责分离。

1、定义异步迭代器方法

using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

public class DataProducer
{
public async IAsyncEnumerable<int> ProduceData(
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
int i = 0;
while (true)
{
cancellationToken.ThrowIfCancellationRequested();
Console.WriteLine($”[Iterator] 生成数字: {i}“);
yield return i++;
await Task.Delay(1000, cancellationToken); // 模拟数据生成延迟
}
}
}

在这个 DataProducer 类中，ProduceData 方法使用 [EnumeratorCancellation] 标注了 cancellationToken 参数。这意味着，当消费者通过 WithCancellation 传递取消令牌时，编译器会自动将该取消令牌传递给 ProduceData 方法的 cancellationToken 参数。

2、定义消费者方法

using System;
using System.Threading;
using System.Threading.Tasks;

public class DataConsumer
{
public async Task ConsumeDataAsync(IAsyncEnumerable<int> producer)
{
using CancellationTokenSource cts = new CancellationTokenSource();

// 在5秒后发出取消请求
_ = Task.Run(async () =>
{
await Task.Delay(5000);
cts.Cancel();
Console.WriteLine(“[Trigger] 已发出取消请求”);
});

try
{
// 通过 WithCancellation 传递取消令牌
await foreach (var data in producer.WithCancellation(cts.Token))
{
Console.WriteLine($”[Consumer] 接收到数据: {data}“);
}
}
catch (OperationCanceledException)
{
Console.WriteLine(“[Consumer] 数据接收已被取消”);
}
}
}

在 DataConsumer 类中，ConsumeDataAsync 方法创建了一个 CancellationTokenSource，并在5秒后取消它。通过 WithCancellation 方法，将取消令牌传递给 ProduceData 方法。这样，消费者完全控制了取消逻辑，而生产者只需响应取消请求。

3、执行示例

public class Program
{
public static async Task Main(string[] args)
{
var producer = new DataProducer();
var consumer = new DataConsumer();
await consumer.ConsumeDataAsync(producer.ProduceData());
}
}

预期输出：

[Iterator] 生成数字: 0
[Consumer] 接收到数据: 0
[Iterator] 生成数字: 1
[Consumer] 接收到数据: 1
[Iterator] 生成数字: 2
[Consumer] 接收到数据: 2
[Iterator] 生成数字: 3
[Consumer] 接收到数据: 3
[Iterator] 生成数字: 4
[Consumer] 接收到数据: 4
[Trigger] 已发出取消请求
[Consumer] 数据接收已被取消

在5秒后，取消请求被触发，迭代器检测到取消并抛出 OperationCanceledException，导致迭代过程被中断。请注意DataConsumer在接收生产出来的数据 IAsyncEnumerable<int> 时，已经错过了在生产函数中传入 cancellationToken 的机会，但作为消费者，仍然可以通过 .WithCancellation 方法进行优雅取消。

这展示了生产者与消费者如何通过 WithCancellation 和 [EnumeratorCancellation] 实现职责分离，消费者能够独立地控制取消逻辑，而生产者只需响应取消请求。

CancellationToken 与 WithCancellation 同时作用时的行为

那么，如果在异步迭代器方法中同时传递了 CancellationToken 参数，并通过 WithCancellation 指定了不同的取消令牌，取消操作会听哪个的？还是都会监听？

结论是：两者都会生效，只要其中任意一个取消令牌被触发，迭代器都会检测到取消请求并中断迭代过程。这取决于方法内部如何处理多个取消令牌。

示例演示

以下是一个详细的示例，展示当同时传递 CancellationToken 参数和使用不同的 WithCancellation 时的行为。

1、定义异步迭代器方法

using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

public class EnumeratorCancellationDemo
{
// 异步迭代器方法，接受两个 CancellationToken
public async IAsyncEnumerable<int> GenerateNumbersAsync(
[EnumeratorCancellation] CancellationToken cancellationToken,
CancellationToken externalCancellationToken = default)
{
int i = 0;
try
{
while (true)
{
// 检查两个取消令牌
cancellationToken.ThrowIfCancellationRequested();
externalCancellationToken.ThrowIfCancellationRequested();

Console.WriteLine($”[Iterator] 生成数字: {i}“);
yield return i++;

// 模拟异步操作
await Task.Delay(1000, cancellationToken);
}
}
finally
{
Console.WriteLine(“[Iterator] 迭代器已退出。”);
}
}
}

2、定义消费者方法

public class Program
{
static async Task Main(string[] args)
{
Console.WriteLine(“启动枚举取消示例…n”);

  var demo = new EnumeratorCancellationDemo();

// 测试1: 先取消方法参数
Console.WriteLine(“=== 测试1: 先取消方法参数 ===n”);
await TestCancellation(demo, cancelParamFirst: true);

// 测试2: 先取消 WithCancellation
Console.WriteLine(“n=== 测试2: 先取消 WithCancellation ===n”);
await TestCancellation(demo, cancelParamFirst: false);

Console.WriteLine(“n演示结束。”);
Console.ReadLine();
}

static async Task TestCancellation(EnumeratorCancellationDemo demo, bool cancelParamFirst)
{
using CancellationTokenSource ctsParam = new CancellationTokenSource();
using CancellationTokenSource ctsWith = new CancellationTokenSource();

if (cancelParamFirst)
{
// 第一个取消任务：3秒后取消 ctsParam
_ = Task.Run(async () =>
{
await Task.Delay(3000);
ctsParam.Cancel();
Console.WriteLine(“[Trigger] 已取消 ctsParam (方法参数)”);
});

// 第二个取消任务：5秒后取消 ctsWith
_ = Task.Run(async () =>
{
await Task.Delay(5000);
ctsWith.Cancel();
Console.WriteLine(“[Trigger] 已取消 ctsWith (WithCancellation)”);
});
}
else
{
// 第一个取消任务：3秒后取消 ctsWith
_ = Task.Run(async () =>
{
await Task.Delay(3000);
ctsWith.Cancel();
Console.WriteLine(“[Trigger] 已取消 ctsWith (WithCancellation)”);
});

// 第二个取消任务：5秒后取消 ctsParam
_ = Task.Run(async () =>
{
await Task.Delay(5000);
ctsParam.Cancel();
Console.WriteLine(“[Trigger] 已取消 ctsParam (方法参数)”);
});
}

try
{
// 传递 ctsWith.Token 作为方法参数，并通过 WithCancellation 传递 ctsWith.Token
await foreach (var number in demo.GenerateNumbersAsync(ctsWith.Token, ctsParam.Token).WithCancellation(ctsWith.Token))
{
Console.WriteLine($”[Consumer] 接收到数字: {number}“);
}
}
catch (OperationCanceledException ex)
{
string reason = ex.CancellationToken == ctsWith.Token ? “WithCancellation” : “方法参数”;
Console.WriteLine($”[Iterator] 迭代器检测到取消。原因: {reason}“);
Console.WriteLine(“[Consumer] 枚举已被取消。”);
}
}
}

3、运行示例并观察结果

启动程序后，控制台输出可能如下所示：

启动枚举取消示例…

=== 测试1: 先取消方法参数 ===

[Iterator] 生成数字: 0
[Consumer] 接收到数字: 0
[Iterator] 生成数字: 1
[Consumer] 接收到数字: 1
[Iterator] 生成数字: 2
[Consumer] 接收到数字: 2
[Trigger] 已取消 ctsParam (方法参数)
[Iterator] 迭代器已退出。
[Iterator] 迭代器检测到取消。原因: 方法参数
[Consumer] 枚举已被取消。

=== 测试2: 先取消 WithCancellation ===

[Iterator] 生成数字: 0
[Consumer] 接收到数字: 0
[Iterator] 生成数字: 1
[Consumer] 接收到数字: 1
[Trigger] 已取消 ctsWith (WithCancellation)
[Iterator] 生成数字: 2
[Consumer] 接收到数字: 2
[Trigger] 已取消 ctsWith (WithCancellation)
[Iterator] 迭代器已退出。
[Iterator] 迭代器检测到取消。原因: WithCancellation
[Consumer] 枚举已被取消。

演示结束。

解释

1、测试1：先取消方法参数 (ctsParam)

在第3秒时，ctsParam 被取消。

迭代器检测到 externalCancellationToken 被取消，抛出 OperationCanceledException。

终止迭代过程，即使 ctsWith 还未被取消。

2、测试2：先取消 WithCancellation (ctsWith)

在第3秒时，ctsWith 被取消。

迭代器检测到 cancellationToken 被取消，抛出 OperationCanceledException。

终止迭代过程，即使 ctsParam 还未被取消。

关键点

• 独立生效：无论是通过方法参数传递的 CancellationToken 还是通过 WithCancellation 传递的 CancellationToken，只要其中一个被取消，迭代器就会响应取消请求并终止迭代。

• 取消顺序无关紧要：不论先取消哪一个取消令牌，迭代器都会正确响应取消请求。取消操作的顺序不会影响最终的效果。

总结

通过上述示例，我们深入了解了 [EnumeratorCancellation] 的必要性及其在异步迭代器中的核心作用。

简要回顾：

消除警告：使用 [EnumeratorCancellation] 可以消除 Visual Studio 提示的警告，确保取消请求能够正确传递给异步迭代器方法。

职责分离：它实现了生产者与消费者的职责分离，使生产者专注于数据生成，消费者控制取消逻辑，从而提升代码的可维护性和可复用性。

灵活的取消机制：即使同时传递多个取消令牌，只要任意一个被取消，迭代器就会终止，提供了灵活而强大的取消控制能力。

.NET 的这些强大功能为开发者提供了极大的便利和灵活性，使得编写高效、可维护的异步代码变得更加轻松与自信。让我们为 .NET 的强大功能自豪，并在实际开发中善加利用这些工具，构建出更优秀的软件解决方案！

最后