一文說通C#中的異步迭代器

今天來寫寫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屬性。而這個屬性表明了前進到的元素。

    爲防止非受權轉發，這兒給出本文的原文連接：https://

 

上面這個例子，咱們經過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。注意，IEnumerable和IEnumerator是不一樣的。
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被取消時，任務會被取消。

總結

同步迭代其實在各個代碼中用的都比較多，但異步迭代用得很好。一方面，這是個相對新的東西，另外一方面，是會有點繞，因此不少人都不敢碰。

今天這個，也是我的的一些經驗總結，但願對你們理解迭代能有所幫助。

 
 

微信公衆號：老王Plus 掃描二維碼，關注我的公衆號，能夠第一時間獲得最新的我的文章和內容推送 本文版權歸做者全部，轉載請保留此聲明和原文連接