當前位置：首頁 > 编程语言 > C# >内容正文

C#

,IAsyncEnumerable 以及搭配异步可能遇到的问题"># C# 重新认识一下 IEnumerable,IAsyncEnumerable 以及搭配异步可能遇到的问题

發布時間：2023/12/24 C# 70 coder

生活随笔收集整理的這篇文章主要介紹了 ,IAsyncEnumerable 以及搭配异步可能遇到的问题"># C# 重新认识一下 IEnumerable,IAsyncEnumerable 以及搭配异步可能遇到的问题小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

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;
    }
}

// 省略部分無關代碼

在調用方 M() 方法中 foreach 關鍵字為我們生成了通過GetTasksFromIEnumerable().GetEnumerator() 方法返回的 IEnumerator<int> 類型的結果的迭代器 ,然后通過try-finally 包裹了原來 forech 中的方法塊 finally 最終會釋放獲取到的迭代器.
GetTasksFromIEnumerable() 方法中為我們生成了一個狀態機 <GetTasksFromIEnumerable>d__1 初始化狀態為 -2 ,然后將當前所處的實例 this 和入參 count 作為字段
通過 <GetTasksFromIEnumerable>d__1 中的 IEnumerable<int>.GetEnumerator() 方法實現該狀態機的初始化,其中還包含了對調用方線程與迭代器初始化線程是否一致的判斷,如果不一致的話會將其重置為當前線程.

然后通過 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 被提前釋放了.

解決錯誤方式很簡單

等待完成 await ParseSimpleTable 后釋放,在當前方法塊中等待完成,但是無法直接返回 IAsyncEnumerable了,必須配合 yield 關鍵字
在最終調用 Stream 的函數中 using 或調用 Close() ,也就是在具體 yield 方法塊之后調用 ,但是在最底層釋放來自調用方的流感覺有點怪怪的(不排除調用方的流還要重用...這里給他關閉了就會顯得坑!)
不偷懶了,手動寫一個基于 string html 解析的函數(哈哈),就沒有上述問題了,也避免了重復創建流對象的問題(滑稽).

總結

在異步中使用一些需要釋放的資源的時候需要注意對象的生命周期,不然可能造成內存泄漏或者代碼異常.
尤其是編寫一些底層一點點的代碼時,往往為了優化而不會同步等待資源到位,而是通過異步的方式訪問,這個時候關注對象的生命周期就顯得尤為重要了.

總結

以上是生活随笔為你收集整理的,IAsyncEnumerable 以及搭配异步可能遇到的问题"># C# 重新认识一下 IEnumerable,IAsyncEnumerable 以及搭配异步可能遇到的问题的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：【已解决】nrm -g安装成功后不是全局
下一篇：小米Civi 3看齐小米13：同款索尼I

C#

,IAsyncEnumerable 以及搭配异步可能遇到的问题"># C# 重新认识一下 IEnumerable,IAsyncEnumerable 以及搭配异步可能遇到的问题

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

前言

什么是IEnumerable<T>

使用方式

1. 通過 GetEnumerator() 方法訪問成員元素

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

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

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