聊一聊C# 8.0中的await foreach

AsyncStreamsInCShaper8.0

很開心今天能與你們一塊兒聊聊C# 8.0中的新特性-Async Streams,通常人一般看到這個詞表情是這樣.

簡單說,其實就是C# 8.0中支持await foreach.

或者說,C# 8.0中支持異步返回枚舉類型async Task<IEnumerable<T>>.

好吧,還不懂?Good,這篇文章就是爲你寫的,看完這篇文章,你就能明白它的神奇之處了.

爲何寫這篇文章

Async Streams這個功能已經發布好久了,在去年的Build 2018 The future of C#就有演示,最近VS 2019發佈,在該版本的Release Notes中,我再次看到了這個新特性,由於對異步編程不太熟悉,因此藉着這個機會,學習新特性的同時,把異步編程重溫一遍.
本文內容,參考了Bassam Alugili在InfoQ中發表的Async Streams in C# 8,撰寫本博客前我已聯繫上該做者並獲得他支持.

Async / Await

C# 5 引入了 Async/Await，用以提升用戶界面響應能力和對 Web 資源的訪問能力。換句話說，異步方法用於執行不阻塞線程並返回一個標量結果的異步操做。

微軟屢次嘗試簡化異步操做，由於 Async/Await 模式易於理解，因此在開發人員當中得到了良好的承認。

詳見The Task asynchronous programming model in C#

常規示例

要了解問什麼須要Async Streams,咱們先來看看這樣的一個示例,求出5之內的整數的和.

static int SumFromOneToCount(int count)
        {
            ConsoleExt.WriteLine("SumFromOneToCount called!");

            var sum = 0;
            for (var i = 0; i <= count; i++)
            {
                sum = sum + i;
            }
            return sum;
        }

調用方法.

static void Main(string[] args)
        {
            const int count = 5;
            ConsoleExt.WriteLine($"Starting the application with count: {count}!");
            ConsoleExt.WriteLine("Classic sum starting.");
            ConsoleExt.WriteLine($"Classic sum result: {SumFromOneToCount(count)}");
            ConsoleExt.WriteLine("Classic sum completed.");
            ConsoleExt.WriteLine("################################################");
        }

輸出結果.

能夠看到,整個過程就一個線程Id爲1的線程自上而下執行,這是最基礎的作法.

Yield Return

接下來,咱們使用yield運算符使得這個方法編程延遲加載,以下所示.

static IEnumerable<int> SumFromOneToCountYield(int count)
        {
            ConsoleExt.WriteLine("SumFromOneToCountYield called!");

            var sum = 0;
            for (var i = 0; i <= count; i++)
            {
                sum = sum + i;

                yield return sum;
            }
        }

主函數

static void Main(string[] args)
        {
            const int count = 5;
            ConsoleExt.WriteLine("Sum with yield starting.");
            foreach (var i in SumFromOneToCountYield(count))
            {
                ConsoleExt.WriteLine($"Yield sum: {i}");
            }
            ConsoleExt.WriteLine("Sum with yield completed.");

            ConsoleExt.WriteLine("################################################");
            ConsoleExt.WriteLine(Environment.NewLine);
        }

運行結果以下.

正如你在輸出窗口中看到的那樣，結果被分紅幾個部分返回，而不是做爲一個值返回。以上顯示的累積結果被稱爲惰性枚舉。可是，仍然存在一個問題，即 sum 方法阻塞了代碼的執行。若是你查看線程ID，能夠看到全部東西都在主線程1中運行,這顯然不完美,繼續改造.

Async Return

咱們試着將async用於SumFromOneToCount方法(沒有yield關鍵字).

static async Task<int> SumFromOneToCountAsync(int count)
        {
            ConsoleExt.WriteLine("SumFromOneToCountAsync called!");

            var result = await Task.Run(() =>
            {
                var sum = 0;

                for (var i = 0; i <= count; i++)
                {
                    sum = sum + i;
                }
                return sum;
            });

            return result;
        }

主函數.

static async Task Main(string[] args)
        {
            const int count = 5;
            ConsoleExt.WriteLine("async example starting.");
            // Sum runs asynchronously! Not enough. We need sum to be async with lazy behavior.
            var result = await SumFromOneToCountAsync(count);
            ConsoleExt.WriteLine("async Result: " + result);
            ConsoleExt.WriteLine("async completed.");

            ConsoleExt.WriteLine("################################################");
            ConsoleExt.WriteLine(Environment.NewLine);
        }

運行結果.

咱們能夠看到計算過程是在另外一個線程中運行，但結果仍然是做爲一個值返回！任然不完美.

若是咱們想把惰性枚舉（yield return）與異步方法結合起來,即返回Task<IEnumerable ,這怎麼實現呢?

Task<IEnumerable >

咱們根據假設把代碼改造一遍,使用Task<IEnumerable<T>>來進行計算.

能夠看到,直接出現錯誤.

IAsyncEnumerable

其實,在C# 8.0中Task<IEnumerable >這種組合稱爲IAsyncEnumerable 。這個新功能爲咱們提供了一種很好的技術來解決拉異步延遲加載的問題，例如從網站下載數據或從文件或數據庫中讀取記錄,與 IEnumerable 和 IEnumerator 相似，Async Streams 提供了兩個新接口 IAsyncEnumerable 和 IAsyncEnumerator ，定義以下：

public interface IAsyncEnumerable<out T>
    {
        IAsyncEnumerator<T> GetAsyncEnumerator();
    }

    public interface IAsyncEnumerator<out T> : IAsyncDisposable
    {
        Task<bool> MoveNextAsync();
        T Current { get; }
    }

   // Async Streams Feature 能夠被異步銷燬 
   public interface IAsyncDisposable
   {
      Task DiskposeAsync();
   }

AsyncStream

下面,咱們就來見識一下AsyncStrema的威力,咱們使用IAsyncEnumerable來對函數進行改造,以下.

static async Task ConsumeAsyncSumSeqeunc(IAsyncEnumerable<int> sequence)
        {
            ConsoleExt.WriteLineAsync("ConsumeAsyncSumSeqeunc Called");

            await foreach (var value in sequence)
            {
                ConsoleExt.WriteLineAsync($"Consuming the value: {value}");

                // simulate some delay!
                await Task.Delay(TimeSpan.FromSeconds(1));
            };
        }

        private static async IAsyncEnumerable<int> ProduceAsyncSumSeqeunc(int count)
        {
            ConsoleExt.WriteLineAsync("ProduceAsyncSumSeqeunc Called");
            var sum = 0;

            for (var i = 0; i <= count; i++)
            {
                sum = sum + i;

                // simulate some delay!
                await Task.Delay(TimeSpan.FromSeconds(0.5));

                yield return sum;
            }
        }

主函數.

static async Task Main(string[] args)
        {
            const int count = 5;
            ConsoleExt.WriteLine("Starting Async Streams Demo!");

            // Start a new task. Used to produce async sequence of data!
            IAsyncEnumerable<int> pullBasedAsyncSequence = ProduceAsyncSumSeqeunc(count);

            // Start another task; Used to consume the async data sequence!
            var consumingTask = Task.Run(() => ConsumeAsyncSumSeqeunc(pullBasedAsyncSequence));

            await Task.Delay(TimeSpan.FromSeconds(3));

            ConsoleExt.WriteLineAsync("X#X#X#X#X#X#X#X#X#X# Doing some other work X#X#X#X#X#X#X#X#X#X#");

            // Just for demo! Wait until the task is finished!
            await consumingTask;

            ConsoleExt.WriteLineAsync("Async Streams Demo Done!");
        }

若是一切順利,那麼就能看到這樣的運行結果了.

最後,看到這就是咱們想要的結果,在枚舉的基礎上,進行了異步迭代.
能夠看到,整個計算過程並無形成主線程的阻塞,其中,值得重點關注的是紅色方框區域的線程5!線程5!線程5!線程5在請求下一個結果後,並無等待結果返回,而是去了Main()函數中作了別的事情,等待請求的結果返回後,線程5又接着執行foreach中任務.

Client/Server的異步拉取

若是尚未理解Async Streams的好處,那麼我藉助客戶端 / 服務器端架構是演示這一功能優點的絕佳方法。

同步調用

客戶端向服務器端發送請求，客戶端必須等待（客戶端被阻塞），直到服務器端作出響應.

示例中Yield Return就是以這種方式執行的,因此整個過程只有一個線程即線程1在處理.

異步調用

客戶端發出數據塊請求，而後繼續執行其餘操做。一旦數據塊到達，客戶端就處理接收到的數據塊並詢問下一個數據塊，依此類推，直到達到最後一個數據塊爲止。這正是 Async Streams 想法的來源。

最後一個示例就是以這種方式執行的,線程5詢問下一個數據後並無等待結果返回,而是去作了Main()函數中的別的事情,數據到達後,線程5又繼續處理foreach中的任務.

Tips

若是你使用的是.net core 2.2及如下版本,會遇到這樣的報錯.

須要安裝.net core 3.0 preview的SDK(截至至博客撰寫日期4月9日,.net core SDK最新版本爲3.0.100-preview3-010431),安裝好SDK後,若是你是VS 2019正式版,可能沒法選擇.net core 3.0,vs 2019 正式版默認狀況下沒有開啓對預覽版.net core 3.0的支持.

根據網友補充,須要在VS 2019正式版本中須要開啓使用 .Net core SDK 預覽版,才能建立3.0的項目.
工具 > 選項 > 項目和解決方案 > .Net Core > 使用 .Net core SDK 預覽版

總結

咱們已經討論過 Async Streams，它是一種出色的異步拉取技術，可用於進行生成多個值的異步計算。

Async Streams 背後的編程概念是異步拉取模型。咱們請求獲取序列的下一個元素，並最終獲得答覆。Async Streams 提供了一種處理異步數據源的絕佳方法，但願對你們可以有所幫助。

文章中涉及的全部代碼已保存在個人GitHub中,請盡情享用!
https://github.com/liuzhenyulive/AsyncStreamsInCShaper8.0

致謝

以前一直感受國外的大師級開發者高不可攀甚至高高在上,在遇到Bassam Alugili以後,我才真正感覺到技術交流沒有高低貴賤,正如他對我說的 The most important thing in this world is sharing the knowledge!
Thank you,I will keep going!!

參考文獻: Async Streams in C# 8 https://www.infoq.com/articles/Async-Streams