C#中的异步迭代器

迭代器的概念

迭代器的概念在C#中出现的比较早,很多人可能已经比较熟悉了。

通常迭代器会用在一些特定的场景中。

举个例子:有一个foreach循环:.

foreach (var item in Sources)
{
    Console.WriteLine(item);
}

这个循环实现了一个简单的功能:把Sources中的每一项在控制台中打印出来。

有时候,Sources可能会是一组完全缓存的数据,例如:List<string>

IEnumerable<string> Sources(int x)
{
    var list = new List<string>();
    for (int i = 0; i < 5; i++)
        list.Add($"result from Sources, x={x}, result {i}");
    return list;
}

这里会有一个小问题:在我们打印Sources的第一个的数据之前,要先运行完整运行Sources()方法来准备数据,在实际应用中,这可能会花费大量时间和内存。更有甚者,Sources可能是一个无边界的列表,或者不定长的开放式列表,比方一次只处理一个数据项目的队列,或者本身没有逻辑结束的队列。

这种情况,C#给出了一个很好的迭代器解决:

IEnumerable<string> Sources(int x)
{
    for (int i = 0; i < 5; i++)
        yield return $"result from Sources, x={x}, result {i}";
}

这个方式的工作原理与上一段代码很像,但有一些根本的区别 - 我们没有用缓存,而只是每次让一个元素可用。

为了帮助理解,来看看foreach在编译器中的解释:

using (var iter = Sources.GetEnumerator())
{
    while (iter.MoveNext())
    {
        var item = iter.Current;
        Console.WriteLine(item);
    }
}

当然,这个是省略掉很多东西后的概念解释,我们不纠结这个细节。但大体的意思是这样的:编译器对传递给foreach的表达式调用GetEnumerator(),然后用一个循环去检查是否有下一个数据(MoveNext()),在得到肯定答案后,前进并访问Current属性。而这个属性代表了前进到的元素。

上面这个例子,我们通过MoveNext()/Current方式访问了一个没有大小限制的向前的列表。我们还用到了yield迭代器这个很复杂的东西 - 至少我是这么认为的。

我们把上面的例子中的yield去掉,改写一下看看:

IEnumerable<string> Sources(int x) => new GeneratedEnumerable(x);
class GeneratedEnumerable : IEnumerable<string>
{
    private int x;
    public GeneratedEnumerable(int x) => this.x = x;
    public IEnumerator<string> GetEnumerator() => new GeneratedEnumerator(x);
    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}
class GeneratedEnumerator : IEnumerator<string>
{
    private int x, i;
    public GeneratedEnumerator(int x) => this.x = x;
    public string Current { get; private set; }
    object IEnumerator.Current => Current;
    public void Dispose() { }
    public bool MoveNext()
    {
        if (i < 5)
        {
            Current = $"result from Sources, x={x}, result {i}";
            i++;
            return true;
        }
        else
        {
            return false;
        }
    }
    void IEnumerator.Reset() => throw new NotSupportedException();
}

这样写完,对照上面的yield迭代器,理解工作过程就比较容易了:

  1. 首先,我们给出一个对象IEnumerable。注意,IEnumerableIEnumerator是不同的。

  2. 当我们调用Sources时,就创建了GeneratedEnumerable。它存储状态参数x,并公开了需要的IEnumerable方法。

  3. 后面,在需要foreach迭代数据时,会调用GetEnumerator(),而它又调用GeneratedEnumerator以充当数据上的游标。

  4. MoveNext()方法逻辑上实现了for循环,只不过,每次调用MoveNext()只执行一步。更多的数据会通过Current回传过来。另外补充一点:MoveNext()方法中的return false对应于yield break关键字,用于终止迭代。

是不是好理解了?

下面说说异步中的迭代器。

异步中的迭代器

上面的迭代,是同步的过程。而现在Dotnet开发工作更倾向于异步,使用async/await来做,特别是在提高服务器的可伸缩性方面应用特别多。

上面的代码最大的问题,在于MoveNext()。很明显,这是个同步的方法。如果它运行需要一段时间,那线程就会被阻塞。这会让代码执行过程变得不可接受。

我们能做得最接近的方法是异步获取数据:

async Task<List<string>> Sources(int x) {...}

但是,异步获取数据并不能解决数据缓存延迟的问题。

好在,C#为此特意增加了对异步迭代器的支持:

public interface IAsyncEnumerable<out T>
{
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
    T Current { get; }
    ValueTask<bool> MoveNextAsync();
}
public interface IAsyncDisposable
{
    ValueTask DisposeAsync();
}

注意,从.NET Standard 2.1.NET Core 3.0开始,异步迭代器已经包含在框架中了。而在早期版本中,需要手动引入:

# dotnet add package Microsoft.Bcl.AsyncInterfaces

目前这个包的版本号是5.0.0。

还是上面例子的逻辑:

IAsyncEnumerable<string> Source(int x) => throw new NotImplementedException();

看看foreach可以await后的样子:

await foreach (var item in Sources)
{
    Console.WriteLine(item);
}

编译器会将它解释为:

await using (var iter = Sources.GetAsyncEnumerator())
{
    while (await iter.MoveNextAsync())
    {
        var item = iter.Current;
        Console.WriteLine(item);
    }
}

这儿有个新东西:await using。与using用法相同,但释放时会调用DisposeAsync,而不是Dispose,包括回收清理也是异步的。

这段代码其实跟前边的同步版本非常相似,只是增加了await。但是,编译器会分解并重写异步状态机,它就变成异步的了。原理不细说了,不是本文关注的内容。

那么,带有yield的迭代器如何异步呢?看代码:

async IAsyncEnumerable<string> Sources(int x)
{
    for (int i = 0; i < 5; i++)
    {
        await Task.Delay(100); // 这儿模拟异步延迟
        yield return $"result from Sources, x={x}, result {i}";
    }
}

嗯,看着就舒服。

这就完了?图样图森破。异步有一个很重要的特性:取消。

那么,怎么取消异步迭代?

异步迭代的取消

异步方法通过CancellationToken来支持取消。异步迭代也不例外。看看上面IAsyncEnumerator<T>的定义,取消标志也被传递到了GetAsyncEnumerator()方法中。

那么,如果是手工循环呢?我们可以这样写:

await foreach (var item in Sources.WithCancellation(cancellationToken).ConfigureAwait(false))
{
    Console.WriteLine(item);
}

这个写法等同于:

var iter = Sources.GetAsyncEnumerator(cancellationToken);
await using (iter.ConfigureAwait(false))
{
    while (await iter.MoveNextAsync().ConfigureAwait(false))
    {
        var item = iter.Current;
        Console.WriteLine(item);
    }
}

没错,ConfigureAwait也适用于DisposeAsync()。所以最后就变成了:

await iter.DisposeAsync().ConfigureAwait(false);

异步迭代的取消捕获做完了,接下来怎么用呢?

看代码:

IAsyncEnumerable<string> Sources(int x) => new SourcesEnumerable(x);
class SourcesEnumerable : IAsyncEnumerable<string>
{
    private int x;
    public SourcesEnumerable(int x) => this.x = x;

    public async IAsyncEnumerator<string> GetAsyncEnumerator(CancellationToken cancellationToken = default)
    {
        for (int i = 0; i < 5; i++)
        {
            await Task.Delay(100, cancellationToken); // 模拟异步延迟
            yield return $"result from Sources, x={x}, result {i}";
        }
    }
}

如果有CancellationToken通过WithCancellation传过来,迭代器会在正确的时间被取消 - 包括异步获取数据期间(例子中的Task.Delay期间)。当然我们还可以在迭代器中任何一个位置检查IsCancellationRequested或调用ThrowIfCancellationRequested()

此外,编译器也会通过[EnumeratorCancellation]来完成这个任务,所以我们还可以这样写:

async IAsyncEnumerable<string> Sources(int x, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    for (int i = 0; i < 5; i++)
    {
        await Task.Delay(100, cancellationToken); // 模拟异步延迟
        yield return $"result from Sources, x={x}, result {i}";
    }
}

这个写法与上面的代码其实是一样的,区别在于加了一个参数。

实际应用中,我们有下面几种写法上的选择:

// 不取消
await foreach (var item in Sources)

// 通过WithCancellation取消
await foreach (var item in Sources.WithCancellation(cancellationToken))

// 通过SourcesAsync取消
await foreach (var item in SourcesAsync(cancellationToken))

// 通过SourcesAsync和WithCancellation取消
await foreach (var item in SourcesAsync(cancellationToken).WithCancellation(cancellationToken))

// 通过不同的Token取消
await foreach (var item in SourcesAsync(tokenA).WithCancellation(tokenB))

几种方式区别于应用场景,实质上没有区别。对两个Token的方式,任何一个Token被取消时,任务会被取消。

总结

同步迭代其实在各个代码中用的都比较多,但异步迭代用得很好。一方面,这是个相对新的东西,另一方面,是会有点绕,所以很多人都不敢碰。