快捷導(dǎo)航

c#使用IAsyncEnumerable實現(xiàn)流式分段傳輸

更新時間：2023年10月20日 16:11:27 作者：四處觀察

這篇文章主要為大家詳細(xì)介紹了c#如何使用IAsyncEnumerable實現(xiàn)流式分段傳輸,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

引言

在使用SSE的時候，前端可以實現(xiàn)流式傳輸，但是有個問題就是這是一個獨(dú)占的連接，相當(dāng)于如果你不手動關(guān)閉連接，就會一直請求，一直連接調(diào)用接口，而且發(fā)送的數(shù)據(jù)格式也是按照定義好的協(xié)議來，而使用c#自帶的IAsyncEnumerable也可以實現(xiàn)流式傳輸，不過返回的數(shù)據(jù)是在之前返回的基礎(chǔ)上進(jìn)行累加，需要自己做處理，我的例子是使用的是ajax來實現(xiàn)，群友有提供了fetch的實現(xiàn)代碼，接下來我們看看c#IAsyncEnumerable實現(xiàn)傳輸?shù)腶jax方案和fetch的代碼吧。

AJAX

下面是源碼和gif效果展示，可以看到我們返回的是一個IAsyncEnumerable<int>類型的結(jié)果，在第二段代碼，我們都知道ajax是根據(jù)xhrhttprequest封裝的，所以自然也可以用一些它的一些事件，所以我們在此處用了onprogress來監(jiān)聽我們請求的進(jìn)度，在這里我們就可以獲取到每一次寫了哪些東西，從而實現(xiàn)一個流傳輸，因為后端寫也是一個字節(jié)一個字節(jié)去寫的，前端接收也是如此。

[HttpGet("Postb")]
 public async IAsyncEnumerable<int> PostB()
 {
    await foreach (var item in  GetData())
     {
         yield return item;
     }
 }
  private async IAsyncEnumerable<int> GetData()
  {
      foreach (int item in Enumerable.Range(0,100))
      {
          await Task.Delay(100);
          yield return item;
      }
  }

<!DOCTYPE html>
<html>
<head>
    <title>AJAX Example</title>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
    <script>

        function callAjax() {
            $.ajax({
                url: 'http://localhost:5203/WeatherForecast/Postb',
    method: 'GET',
    contentType: 'application/json',
    xhrFields: {
      onprogress: function (e) {
        var msg = e.currentTarget.response;
        $("#list").append(`<h5>${msg}</h5>`);
        console.log("接收的數(shù)據(jù)是=>" + msg);
        },
        onchange: function (a) {
            debugger;
        }
    },
    success: function () {
      console.log("分塊讀取完成");
    },
    error: function (xhr, status, error) {
      console.log("請求失敗");
      console.log("錯誤信息: " + error);
    }
  });
        }
    </script>
</head>
<body>
    <button onclick="callAjax()">調(diào)用AJAX</button>
</body>
</html>

SSE

SSE全稱Server Sent Event，從名字我們可以看出，這是一個服務(wù)端單向發(fā)送到客戶端的，與WebSocket不同，但是兩者都是長連接，上面的ajax的響應(yīng)標(biāo)頭是applycation/json，SSE的必須是text/event-stream,并且SSE的發(fā)送的參數(shù)也都是有固定的格式，每一個發(fā)送的消息都是由\n\n分割，每一個message由若干個可選的字段組成，例如下面，field：value是一個message里面的內(nèi)容，field可選范圍是下面那四個，第二代碼段是后端的代碼，展示了一個完整的message，包括了data，event，retry和id，其中上面，我們設(shè)置了響應(yīng)的Content-type是text/event-stream，設(shè)置是不緩存no-cache，下面設(shè)置是保持連接，keepalive，因為是長連接嘛，id和data可以隨便給，retry是端口連接后的一個重新連接時間，event是一個事件的名稱，我們給客戶端返回這個格式的內(nèi)容，客戶端就會根據(jù)這個內(nèi)容就返回數(shù)據(jù)，調(diào)用我們的event，從而實現(xiàn)一個流式輸出。

[field]: value\n  //這是一個Message

//下面是可選的字段
data
event
id
retry

[HttpGet("Posta")]
 public  IActionResult Posta()
 {
     if (Response.Headers.ContainsKey("Content-Type"))
     {
         Response.Headers.Remove("Content-Type");
         Response.Headers.Add("Content-Type", "text/event-stream");
     }
     else
     {
         Response.Headers.Add("Content-Type", "text/event-stream");
     }
     Response.Headers.Add("Cache-Control", "no-cache");
     Response.Headers.Add("Connection", "keep-alive");
     string data =
               $"id: {Random.Shared.Next()} \n" +
               $"retry: {Random.Shared.Next(0, 100) * 30}\n" +
               $"event: message\n" +
               $"data: {Random.Shared.Next()}\n\n";return Content(data);
 }

<!DOCTYPE html>
<html>
<head>
    <title>SSE Example</title>
    <script>
        var eventSource = new EventSource("http://localhost:5203/WeatherForecast/Posta");  
        eventSource.addEventListener("message", function(event) {
var a=document.getElementById("aaa");
a.innerHTML+="<a>"+event.data+"</a><br>"
            console.log("Received message: " + event.data);
        });
        eventSource.addEventListener("error", function(event) {
            console.log("Error occurred");
        });
    </script>
</head>
<body>
<div id='aaa'></div>
</body>
</html>