很開心今天能與大家一起聊聊C# 8.0中的新特性-Async Streams,一般人通常看到這個(gè)詞表情是這樣.

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

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

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

為什么寫這篇文章

Async Streams這個(gè)功能已經(jīng)發(fā)布很久了,在去年的Build 2018 The future of C#就有演示,最近VS 2019發(fā)布,在該版本的Release Notes中,我再次看到了這個(gè)新特性,因?yàn)閷Ξ惒骄幊滩惶煜?所以借著這個(gè)機(jī)會,學(xué)習(xí)新特性的同時(shí),把異步編程重溫一遍.
本文內(nèi)容,參考了Bassam Alugili在InfoQ中發(fā)表的Async Streams in C# 8,撰寫本博客前我已聯(lián)系上該作者并得到他支持.

Async / Await

C#?5?引入了?Async/Await，用以提高用戶界面響應(yīng)能力和對?Web?資源的訪問能力。換句話說，異步方法用于執(zhí)行不阻塞線程并返回一個(gè)標(biāo)量結(jié)果的異步操作。

微軟多次嘗試簡化異步操作，因?yàn)?Async/Await?模式易于理解，所以在開發(fā)人員當(dāng)中獲得了良好的認(rèn)可。

詳見The Task asynchronous programming model in C#

常規(guī)示例

要了解問什么需要Async Streams,我們先來看看這樣的一個(gè)示例,求出5以內(nèi)的整數(shù)的和.

Copy

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

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

調(diào)用方法.

Copy

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("################################################");
}

輸出結(jié)果.

可以看到,整個(gè)過程就一個(gè)線程Id為1的線程自上而下執(zhí)行,這是最基礎(chǔ)的做法.

Yield Return

接下來,我們使用yield運(yùn)算符使得這個(gè)方法編程延遲加載,如下所示.

Copy

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

主函數(shù)

Copy

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

運(yùn)行結(jié)果如下.

正如你在輸出窗口中看到的那樣，結(jié)果被分成幾個(gè)部分返回，而不是作為一個(gè)值返回。以上顯示的累積結(jié)果被稱為惰性枚舉。但是，仍然存在一個(gè)問題，即?sum?方法阻塞了代碼的執(zhí)行。如果你查看線程ID，可以看到所有東西都在主線程1中運(yùn)行,這顯然不完美,繼續(xù)改造.

Async Return

我們試著將async用于SumFromOneToCount方法(沒有yield關(guān)鍵字).

Copy

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

主函數(shù).

Copy

static async Task Main(string[] args)
{
const int count = 5;
ConsoleExt.WriteLine("async example starting.");

var result = await SumFromOneToCountAsync(count);
ConsoleExt.WriteLine("async?Result:?" + result);
ConsoleExt.WriteLine("async completed.");

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

運(yùn)行結(jié)果.

我們可以看到計(jì)算過程是在另一個(gè)線程中運(yùn)行，但結(jié)果仍然是作為一個(gè)值返回！任然不完美.

如果我們想把惰性枚舉（yield return）與異步方法結(jié)合起來,即返回Task<IEnumerable,這怎么實(shí)現(xiàn)呢?

Task<IEnumerable>

我們根據(jù)假設(shè)把代碼改造一遍,使用Task<IEnumerable<T>>來進(jìn)行計(jì)算.

可以看到,直接出現(xiàn)錯(cuò)誤.

IAsyncEnumerable

其實(shí),在C# 8.0中Task<IEnumerable>這種組合稱為IAsyncEnumerable。這個(gè)新功能為我們提供了一種很好的技術(shù)來解決拉異步延遲加載的問題，例如從網(wǎng)站下載數(shù)據(jù)或從文件或數(shù)據(jù)庫中讀取記錄,與?IEnumerable?和 IEnumerator?類似，Async Streams 提供了兩個(gè)新接口 IAsyncEnumerable?和 IAsyncEnumerator，定義如下：

Copy

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

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


public interface IAsyncDisposable
{
Task DiskposeAsync();
}

AsyncStream

下面,我們就來見識一下AsyncStrema的威力,我們使用IAsyncEnumerable來對函數(shù)進(jìn)行改造,如下.

Copy

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

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


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;


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

yield return sum;
}
}

主函數(shù).

Copy

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


IAsyncEnumerable<int> pullBasedAsyncSequence = ProduceAsyncSumSeqeunc(count);


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#");


await consumingTask;

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

如果一切順利,那么就能看到這樣的運(yùn)行結(jié)果了.

最后,看到這就是我們想要的結(jié)果,在枚舉的基礎(chǔ)上,進(jìn)行了異步迭代.
可以看到,整個(gè)計(jì)算過程并沒有造成主線程的阻塞,其中,值得重點(diǎn)關(guān)注的是紅色方框區(qū)域的線程5!線程5!線程5!線程5在請求下一個(gè)結(jié)果后,并沒有等待結(jié)果返回,而是去了Main()函數(shù)中做了別的事情,等待請求的結(jié)果返回后,線程5又接著執(zhí)行foreach中任務(wù).

Client/Server的異步拉取

如果還沒有理解Async Streams的好處,那么我借助客戶端?/?服務(wù)器端架構(gòu)是演示這一功能優(yōu)勢的絕佳方法。

同步調(diào)用

客戶端向服務(wù)器端發(fā)送請求，客戶端必須等待（客戶端被阻塞），直到服務(wù)器端做出響應(yīng).

示例中Yield Return就是以這種方式執(zhí)行的,所以整個(gè)過程只有一個(gè)線程即線程1在處理.

異步調(diào)用

客戶端發(fā)出數(shù)據(jù)塊請求，然后繼續(xù)執(zhí)行其他操作。一旦數(shù)據(jù)塊到達(dá)，客戶端就處理接收到的數(shù)據(jù)塊并詢問下一個(gè)數(shù)據(jù)塊，依此類推，直到達(dá)到最后一個(gè)數(shù)據(jù)塊為止。這正是?Async?Streams?想法的來源。

最后一個(gè)示例就是以這種方式執(zhí)行的,線程5詢問下一個(gè)數(shù)據(jù)后并沒有等待結(jié)果返回,而是去做了Main()函數(shù)中的別的事情,數(shù)據(jù)到達(dá)后,線程5又繼續(xù)處理foreach中的任務(wù).

Tips

如果你使用的是.net core 2.2及以下版本,會遇到這樣的報(bào)錯(cuò).

需要安裝.net core 3.0 preview的SDK(截至至博客撰寫日期4月9日,.net core SDK最新版本為3.0.100-preview3-010431),安裝好SDK后,如果你是VS 2019正式版,可能無法選擇3.0的與預(yù)覽版,聽過只有VS 2019 Preview才支持.Net core 3.0的預(yù)覽版.

總結(jié)

我們已經(jīng)討論過?Async Streams，它是一種出色的異步拉取技術(shù)，可用于進(jìn)行生成多個(gè)值的異步計(jì)算。

Async Streams?背后的編程概念是異步拉取模型。我們請求獲取序列的下一個(gè)元素，并最終得到答復(fù)。Async?Streams?提供了一種處理異步數(shù)據(jù)源的絕佳方法，希望對大家能夠有所幫助。

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

致謝

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

參考文獻(xiàn):?Async?Streams?in?C#?8?https://www.infoq.com/articles/Async-Streams

原文地址:?https://www.cnblogs.com/CoderAyu/p/10680805.html

.NET社區(qū)新聞，深度好文，歡迎訪問公眾號文章匯總?http://www.csharpkit.com?

總結(jié)

以上是生活随笔為你收集整理的聊一聊C# 8.0中的await foreach的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。