技術文章 » # C# 重新認識一下 IEnumerable<T>,IAsyncEnumerable<T> 以及搭配非同步可能遇到的問題

# C# 重新認識一下 IEnumerable<T>,IAsyncEnumerable<T> 以及搭配非同步可能遇到的問題

2023-12-14 18:00:50

C# 重新認識一下 IEnumerable<T>,IAsyncEnumerable<T> 以及搭配非同步可能遇到的問題

前言

為啥會想到寫這個
為了這碟醋,包了這頓餃子
作為老鳥不免犯迷糊
因為 在使用非同步中使用IEnumerable<T>,IAsyncEnumerable<T>遇到了一些細節(對於我之前來說)上沒注意到問題.

什麼是IEnumerable<T>

IEnumerable<T> 繼承自 System.Collections.IEnumerable


namespace System.Collections.Generic
{
    //
    // 摘要:
    //     Exposes the enumerator, which supports a simple iteration over a collection of
    //     a specified type.
    //
    // 型別引數:
    //   T:
    //     The type of objects to enumerate.
    public interface IEnumerable<out T> : IEnumerable
    {
        //
        // 摘要:
        //     Returns an enumerator that iterates through the collection.
        //
        // 返回結果:
        //     An enumerator that can be used to iterate through the collection.
        IEnumerator<T> GetEnumerator();
    }
}

以下參照自 微軟官方檔案

IEnumerable<T>是 名稱空間中System.Collections.Generic集合（例如 、 Dictionary<TKey,TValue>和 Stack<T> ）List<T>和其他泛型集合（如 ObservableCollection<T> 和 ConcurrentStack<T>）的基介面。 可以使用 語句列舉實現 IEnumerable<T> 的 foreach 集合。

有關此介面的非泛型版本，請參閱 System.Collections.IEnumerable。

IEnumerable<T> 包含實現此介面時必須實現的單個方法; GetEnumerator，返回 IEnumerator<T> 物件。 返回的 IEnumerator<T> 提供通過公開 Current 屬性迴圈存取集合的功能。

粗俗的說,就是我們可以通過實現了 IEnumerable<T> 介面的容器提高資料處理的效率,因為通過它 我們可以方便的使用 foreach 關鍵字 遍歷容器內的元素,而我們所熟知的大部分的容器,例如,List<T>,Dictionary<TKey,TValue> 等等都是實現了 IEnumerable<T> 的.

除了快速遍歷以外,作為返回值 IEnumerable<T> 也有著強大的優勢,因為如果是傳統的陣列遍歷的話如果我想要找到多個陣列中指定的元素,我必須等到找到所有符合的元素的時候才能將資料返回,呼叫方才能開始進行操作,而返回結果為 IEnumerable<T> 的方法可以通過 yield 關鍵字提前將當前符合條件的 T 值返回給呼叫方然後返回到之前執行的地方繼續查詢符合條件的元素.

使用方式

1. 通過 GetEnumerator() 方法存取成員元素

IEnumerable和IEnumerable<T>介面提供了GetEnumerator()方法讓我們獲取迭代器,通過MoveNext()方法返回的bool值提供是否可以進行下一次迭代,然後通過Current屬性獲取當前元素.


    // 快速生成0-100, Enumerable 提供了很多方便的靜態方法
    IEnumerable<int> arr = Enumerable.Range(0, 100);

    var enumerator = arr.GetEnumerator();

    while(enumerator.MoveNext())
    {
        enumerator.Current.Dump();
    }

2. 通過 foreach 關鍵字快速遍歷成員元素

foreach關鍵字提供了快速遍歷成員元素的操作,其也是通過生成第一個例子的程式碼迭代,省去了反覆書寫冗餘程式碼的步驟.

微軟官方建議使用 foreach，而不是直接操作列舉數

(這裡是一個鴨子型別)只要擁有GetEnumerator方法都可以通過foreach關鍵字進行遍歷,所可以通過一些黑魔法(擴充套件函數Range型別範例GetEnumerator)實現 foreach (var i in 1..10) 這樣的語法.


    IEnumerable<int> arr = Enumerable.Range(0, 100);    
    
    // 遍歷列印成員
    foreach (int element in arr)
    {
        Console.WriteLine(arr.ToString());  
    }

3. 作為同步方法返回值時通過 yield 關鍵字即時返回成員

當使用IEnumerable<T>作為同步方法的返回值時,我們可以對外隱藏返回值具體的實現,比如List<T> 實現了IEnumerable<T>,Dictionary<TKey,TValue>實現了IEnumerable<KeyValuePair<TKey,TValue>>.

當需要返回值時,方法內可以是一個整體結果返回,也可以利用yield關鍵字逐個成員結果返回.


    public void Main(string[] args)
    {
        // 通過IEnumerable<char> 逐個char 列印
        foreach (var task in GetTasksFromIEnumerable(5))
        {
            Console.WriteLine(task);
            Console.WriteLine($"處理完:{task}");
        }

        IEnumerable<int> GetTasksFromIEnumerable(int count)
        {
            for (int i = 0; i < count; i++)
            {
                yield return HeavyTask(i);
                Console.WriteLine($"已返回當前值:{i},準備下一次");
            }
        }

        // 模擬比較重的任務
        int HeavyTask(int i)
        {
            // 模擬耗時
            Thread.Sleep(1000);

            return i;
        }
    }

以上程式碼我們可以得到以下輸出,可以看到每次呼叫方當前迴圈體結束後,迭代器又會回到當前執行的地方準備執行下一次迭代;

0
處理完:0
已返回當前值:0,準備下一次
1
處理完:1
已返回當前值:1,準備下一次
2
處理完:2
已返回當前值:2,準備下一次
3
處理完:3
已返回當前值:3,準備下一次
4
處理完:4
已返回當前值:4,準備下一次

4. 作為非同步方法返回值時通過 yield 關鍵字即時返回成員

在如今非同步方法大行其道的今天,我們的實際使用中非同步方法已經稀疏平常了,但 C# 中的非同步方法關鍵字 async , await 具有傳染性,只有我們方法中使用到了非同步方法並希望使用 await 等待結果的時候當前的方法必須使用 async 關鍵字標記並且將返回值使用 Task<T> 包裹.所以,通過正常途徑我們無法獲得一個只返回 IEnumerable<T> 結果的非同步方法,因為它始終被 Task 包裹,除非我們在方法中等待所有的結果完成後作為非同步方法的結果返回,但顯然這不是我們希望的結果.那麼我們如何才能希望和同步方法中一樣即時返回當前的結果且不阻塞呢? 答案是使用它的非同步型別介面 IAsyncEnumerable<T>.

可以使整個結果返回,無法將單個結果即時返回

    public async Task<IEnumerable<int>> GetNumbersAsync()
    {
        // 模擬需要執行的非同步任務
        await Task.Delay(1000);

        var result = Enumerable.Range(0, 100);

        return result; //  ✔ 返回整個結果
    }

    public async Task<IEnumerable<int>> GetNumbersAsync()
    {
        for(int i = 0; i < 5 ; i++ )
        {
            yield return await GetSignleNumberAsync(); //  ❌ 編譯錯誤 

            //CS1624: The body of 'GetNumbersAsync()' cannot be an iterator block because 'Task<IEnumerable<int>>' is not an iterator interface type
        }
    }

5. IAsyncEnumerable<T>

當使用 IAsyncEnumerable<T> 時非同步方法的返回值可以直接使用它作為返回值的型別例如


    public async Task Main(string[]args)
    {
        Console.WriteLine($"當前執行緒:{Environment.CurrentManagedThreadId}");

        // 通過await foreach 立即進行迭代
        await foreach (var number in GetNumbersAsync())
        {
            Console.WriteLine($"當前執行緒:{Environment.CurrentManagedThreadId}");
            Console.WriteLine(number);
        }
    }

    async IAsyncEnumerable<int> GetNumbersAsync()
    {
        for (int i = 0; i < 5; i++)
        {
            yield return await GetSignleNumberAsync(); //  ✔ 編譯通過
        }
    }

    async Task<int> GetSignleNumberAsync()
    {
        // 模擬耗時
        await Task.Delay(1000);

        return Random.Shared.Next();
    }

得到輸出結果

當前執行緒:1
當前執行緒:6
809282356
當前執行緒:6
696341357
當前執行緒:6
872147671
當前執行緒:6
791323674
當前執行緒:6
1961595625
當前執行緒:6

我們也可以通過 ToBlockingEnumerable() 方法將對應的 IAsyncEnumerable<int> 的結果轉為同步阻塞的 IEnumerable<T>


// 通過 ToBlockingEnumerable 轉為同步阻塞的 IEnumerable<T>
var result = GetNumbersAsync().ToBlockingEnumerable();

// 將以同步程式碼執行
Console.WriteLine($"當前執行緒:{Environment.CurrentManagedThreadId}");
foreach (var element in result)
{ 
    Console.WriteLine($"當前執行緒:{Environment.CurrentManagedThreadId}");
    Console.WriteLine(element);
}

得到以下輸出結果

當前執行緒:1
當前執行緒:1
1933649614
當前執行緒:1
1975509029
當前執行緒:1
1303323564
當前執行緒:1
1618007076
當前執行緒:1
503278324

IEnumerable 到底做了什麼

我們可以通過 sharplab.io 這個網站來看看 通過 yield + foreach 關鍵字為我們生成最終的程式碼的樣子

原始碼

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

public class C 
{
    public void M() 
    {
        foreach(var item in GetTasksFromIEnumerable(15))
        {
            Console.WriteLine(item);
        }
    }
    
    IEnumerable<int> GetTasksFromIEnumerable(int count)
    {
        for (int i = 0; i < count; i++)
        {
            yield return HeavyTask(i);
            Console.WriteLine($"已返回當前值:{i},準備下一次");
        }
    }

    // 模擬比較重的任務
    int HeavyTask(int i)
    {
        // 模擬耗時
        Thread.Sleep(1000);

        return i;
    }
}

生成後的程式碼


// 省略部分無關程式碼 
public class C
{
    [CompilerGenerated]
    private sealed class <GetTasksFromIEnumerable>d__1 : IEnumerable<int>, IEnumerable, IEnumerator<int>, IEnumerator, IDisposable
    {
        private int <>1__state;

        private int <>2__current;

        private int <>l__initialThreadId;

        private int count;

        public int <>3__count;

        public C <>4__this;

        private int <i>5__1;

        int IEnumerator<int>.Current
        {
            [DebuggerHidden]
            get
            {
                return <>2__current;
            }
        }

        object IEnumerator.Current
        {
            [DebuggerHidden]
            get
            {
                return <>2__current;
            }
        }

        [DebuggerHidden]
        public <GetTasksFromIEnumerable>d__1(int <>1__state)
        {
            this.<>1__state = <>1__state;
            <>l__initialThreadId = Environment.CurrentManagedThreadId;
        }

        [DebuggerHidden]
        void IDisposable.Dispose()
        {
        }

        private bool MoveNext()
        {
            int num = <>1__state;
            if (num != 0)
            {
                if (num != 1)
                {
                    return false;
                }
                <>1__state = -1;
                DefaultInterpolatedStringHandler defaultInterpolatedStringHandler = new DefaultInterpolatedStringHandler(13, 1);
                defaultInterpolatedStringHandler.AppendLiteral("已返回當前值:");
                defaultInterpolatedStringHandler.AppendFormatted(<i>5__1);
                defaultInterpolatedStringHandler.AppendLiteral(",準備下一次");
                Console.WriteLine(defaultInterpolatedStringHandler.ToStringAndClear());
                <i>5__1++;
            }
            else
            {
                <>1__state = -1;
                <i>5__1 = 0;
            }
            if (<i>5__1 < count)
            {
                <>2__current = <>4__this.HeavyTask(<i>5__1);
                <>1__state = 1;
                return true;
            }
            return false;
        }

        bool IEnumerator.MoveNext()
        {
            //ILSpy generated this explicit interface implementation from .override directive in MoveNext
            return this.MoveNext();
        }

        [DebuggerHidden]
        void IEnumerator.Reset()
        {
            throw new NotSupportedException();
        }

        [DebuggerHidden]
        [return: System.Runtime.CompilerServices.Nullable(1)]
        IEnumerator<int> IEnumerable<int>.GetEnumerator()
        {
            <GetTasksFromIEnumerable>d__1 <GetTasksFromIEnumerable>d__;
            if (<>1__state == -2 && <>l__initialThreadId == Environment.CurrentManagedThreadId)
            {
                <>1__state = 0;
                <GetTasksFromIEnumerable>d__ = this;
            }
            else
            {
                <GetTasksFromIEnumerable>d__ = new <GetTasksFromIEnumerable>d__1(0);
                <GetTasksFromIEnumerable>d__.<>4__this = <>4__this;
            }
            <GetTasksFromIEnumerable>d__.count = <>3__count;
            return <GetTasksFromIEnumerable>d__;
        }

        [DebuggerHidden]
        [return: System.Runtime.CompilerServices.Nullable(1)]
        IEnumerator IEnumerable.GetEnumerator()
        {
            return ((IEnumerable<int>)this).GetEnumerator();
        }
    }

    public void M()
    {
        IEnumerator<int> enumerator = GetTasksFromIEnumerable(15).GetEnumerator();
        try
        {
            while (enumerator.MoveNext())
            {
                int current = enumerator.Current;
                Console.WriteLine(current);
            }
        }
        finally
        {
            if (enumerator != null)
            {
                enumerator.Dispose();
            }
        }
    }

    [System.Runtime.CompilerServices.NullableContext(1)]
    [IteratorStateMachine(typeof(<GetTasksFromIEnumerable>d__1))]
    private IEnumerable<int> GetTasksFromIEnumerable(int count)
    {
        <GetTasksFromIEnumerable>d__1 <GetTasksFromIEnumerable>d__ = new <GetTasksFromIEnumerable>d__1(-2);
        <GetTasksFromIEnumerable>d__.<>4__this = this;
        <GetTasksFromIEnumerable>d__.<>3__count = count;
        return <GetTasksFromIEnumerable>d__;
    }

    private int HeavyTask(int i)
    {
        Thread.Sleep(1000);
        return i;
    }
}

// 省略部分無關程式碼

  1. 在 呼叫方 M() 方法中 foreach 關鍵字 為我們生成了通過GetTasksFromIEnumerable().GetEnumerator() 方法返回的 IEnumerator<int> 型別的結果 的迭代器 ,然後通過try-finally 包裹了原來 forech 中的方法塊 finally 最終會釋放獲取到的迭代器.

  2. GetTasksFromIEnumerable() 方法中為我們生成了一個狀態機 <GetTasksFromIEnumerable>d__1 初始化狀態為 -2 ,然後將 當前所處的範例 this 和 入參 count 作為欄位

  3. 通過 <GetTasksFromIEnumerable>d__1 中的 IEnumerable<int>.GetEnumerator() 方法實現該狀態機的初始化,其中還包含了對呼叫方執行緒與迭代器初始化執行緒是否一致的判斷,如果不一致的話會將其重置為當前執行緒.

  4. 然後通過 MoveNext 不斷獲取當前迭代的值 ,可以看到原來的

    yield return HeavyTask(i);

    轉化成了

     if (<i>5__1 < count) // 原來條件
 {
     <>2__current = <>4__this.HeavyTask(<i>5__1);
     <>1__state = 1; // 將 state 標記為 1, 使其走到上面對應的 if 語句
     return true; // 並表示可以繼續移動
 }
 return false; // 結束

    state 改變為 1 之後 , 執行原 yield 後的程式碼塊

    if (num != 0)
{

    if (num != 1)
    {
        return false;
    }

    // 重新標記為 -1
    <>1__state = -1;

    // 對應原來的 Console.WriteLine($"已返回當前值:{i},準備下一次");
    DefaultInterpolatedStringHandler defaultInterpolatedStringHandler = new DefaultInterpolatedStringHandler(13, 1);
    defaultInterpolatedStringHandler.AppendLiteral("已返回當前值:");
    defaultInterpolatedStringHandler.AppendFormatted(<i>5__1);
    defaultInterpolatedStringHandler.AppendLiteral(",準備下一次");
    Console.WriteLine(defaultInterpolatedStringHandler.ToStringAndClear());
    
    // 迴圈遍歷累加
    <i>5__1++;
}
else
{
    <>1__state = -1;
    // 這裡為啥會重置為 0 ?
    <i>5__1 = 0;
}

問題

上面說了為了這碟醋包了這頓餃子,那麼這頓餃子是什麼呢?

其實後面發現不是 IEnumerable 或者IAsyncEnumerable 的問題 而是對於非同步中物件的生命週期的理解問題.

之前再寫一個解析網頁元素項的輔助方法時,本著能少寫一個少寫一個的原則(哈哈哈,偷懶),想將傳入的 html 字串轉成流 然後呼叫另一個寫好的 Stream 解析的函數.


/// 偷懶的函數
public static IAsyncEnumerable<TTableRow> ParseSimpleTable<TTableRow>(string html, string tableSelector, string rowSelector, Func<IElement, ValueTask<TTableRow>> rowParseFunc)
{
    // 出於直覺 在這裡 using 
    using MemoryStream stream = new MemoryStream(Encoding.UTF8.GetBytes(html));

    return ParseSimpleTable(stream, tableSelector, rowSelector, rowParseFunc);
}

/// <summary>
/// 解析簡單表格
/// </summary>
/// <typeparam name="TTableRow">解析結果項</typeparam>
/// <param name="stream">要解析的流</param>
/// <param name="tableSelector">table選擇器</param>
/// <param name="rowSelector">行選擇器</param>
/// <param name="rowParseFunc">行解析方法委託</param>
/// <returns></returns>
/// <exception cref="ArgumentException"></exception>
public static async IAsyncEnumerable<TTableRow> ParseSimpleTable<TTableRow>(Stream stream, string tableSelector, string rowSelector, Func<IElement, ValueTask<TTableRow>> rowParseFunc)
{
    IBrowsingContext browsingContext = BrowsingContext.New();

    var htmlParser = browsingContext.GetService<IHtmlParser>();

    if (htmlParser == null)
        throw new ArgumentException(nameof(htmlParser));

    using IDocument document = await htmlParser.ParseDocumentAsync(stream);

    var tableElement = document.QuerySelector(tableSelector);
    if (tableElement == null)
        yield break;

    var rowsElement = tableElement.QuerySelectorAll(rowSelector);
    if (rowsElement == null || !rowsElement.Any())
        yield break;

    foreach (var rowElement in rowsElement)
    {
        yield return await rowParseFunc(rowElement);
    }
}

由於出於直覺的 using 了這個流,下意識的以為這個 Stream 會在這個函數執行後釋放, 然後就...異常了

Cannot access a closed Stream.
  Data = <enumerable Count: 0>
  HelpLink = <null>
  HResult = -2146232798
  InnerException = <null>
  Message = Cannot access a closed Stream.
  ObjectName = 
  Source = System.Private.CoreLib
  StackTrace =    at System.IO.MemoryStream.get_Length()
   at Program.<<Main>$>g__GetBytes|0_1(Stream stream)+MoveNext() in :line 20
   at Program.<<Main>$>g__GetBytes|0_1(Stream stream)+System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult()
   at Program.<Main>$(String[] args) in :line 3
   at Program.<Main>$(String[] args) in :line 3
   at Program.<Main>(String[] args)
  TargetSite = Void ThrowObjectDisposedException_StreamClosed(System.String)

一般流報這個異常都是被提前釋放的問題,我一想噢應該時非同步的問題,然後我去看生成後的程式碼,恍然大悟.


// 模擬場景 

 private IAsyncEnumerable<byte> ParseSimpleTable<TTableRow>(string s)
    {
        MemoryStream memoryStream = new MemoryStream(Encoding.UTF8.GetBytes(s));
        try
        {
            // 這裡是一個非同步方法,但是我並沒有等待完成,而是轉交給了呼叫方等待
            return ParseSimpleTable(memoryStream);
        }
        finally
        {
            if (memoryStream != null)
            {
                // 沒有等待所以這裡 memoryStream 被釋放了 ,但是 GetBytes 方法還在執行
                ((IDisposable)memoryStream).Dispose();
            }
        }
    }

生成後的程式碼 一目瞭然,memoryStream 被提前釋放了.

解決錯誤方式很簡單

  1. 等待完成 await ParseSimpleTable 後釋放,在當前方法塊中等待完成,但是無法直接返回 IAsyncEnumerable了,必須配合 yield 關鍵字

  2. 在最終呼叫 Stream 的函數中 using 或 呼叫 Close() ,也就是在具體 yield 方法塊之後呼叫 ,但是在最底層釋放來自呼叫方的流感覺有點怪怪的(不排除呼叫方的流還要重用...這裡給他關閉了就會顯得坑!)

  3. 不偷懶了,手動寫一個 基於 string html 解析的函數(哈哈),就沒有上述問題了,也避免了重複建立流物件的問題(滑稽).

總結

在非同步中使用一些需要釋放的資源的時候需要注意物件的生命週期,不然可能造成記憶體漏失或者程式碼異常.
尤其是編寫一些底層一點點的程式碼時,往往為了優化而不會同步等待資源到位,而是通過非同步的方式存取,這個時候關注物件的生命週期就顯得尤為重要了.