基于HTTP2/3的流模式消息交換如何實現？
當前位置：點晴教程→知識管理交流
→『 技術文檔交流 』
我想很多人已經體驗過GRPC提供的三種流式消息交換（Client Stream、Server Stream和Duplex Stream）模式，在.NET Core上構建的GRPC應用本質上是采用HTTP2/HTTP3協議的ASP.NET Core應用，我們當然也可以在一個普通的ASP.NET Core應用實現這些流模式。不僅如此，HttpClient也提供了響應的支持，這篇文章通過一個簡單的實例提供了相應的實現，源代碼從這里下載。
一、雙向流的效果在提供具體實現之前，我們不妨先來演示一下最終的效果。我們通過下面這段代碼構建了一個簡單的ASP.NET Core應用，如代碼片段所示，在調用WebApplication的靜態方法CreateBuilder將WebApplicationBuilder創建出來后，我們調用其擴展方法UseKestrel將默認終結點的監聽協議設置為Http1AndHttp2AndHttp3，這樣我們的應用將提供針對不同HTTP協議的全面支持。 var url = "http://localhost:9999";
var builder = WebApplication.CreateBuilder(args);
builder.WebHost
.UseKestrel(kestrel=> kestrel.ConfigureEndpointDefaults(listen=>listen.Protocols = HttpProtocols.Http1AndHttp2AndHttp3))
.UseUrls(url);
var app = builder.Build();
app.MapPost("/", httpContext=> HandleRequestAsync(httpContext, async (request, writer) => {
Console.WriteLine($"[Server]Receive request message: {request}");
await writer.WriteStringAsync(request);
}));
await app.StartAsync();
await SendStreamRequestAsync(url, ["foo", "bar", "baz", "qux"], reply => {
Console.WriteLine($"[Client]Receive reply message: {reply}\n");
return Task.CompletedTask;
});我們針對根路徑（/）注冊了一個HTTP方法為POST的路由終結點，終結點處理器調用HanleRequestAsync來處理請求。這個方法提供一個Func<string， PipeWriter, Task>類型的參數作為處理器，該委托的第一個參數表示接收到的單條請求消息，PipeWriter用來寫入響應內容。在這里我們將接收到的消息進行簡單格式化后將其輸出到控制臺上，隨之將其作為響應內容進行回寫。 在應用啟動之后，我們調用SendStreamRequestAsync方法以流的方式發送請求，并處理接收到的響應內容。該方法的第一個參數為請求發送的目標URL，第二個參數是一個字符串數組，我們將以流的方式逐個發送每個字符串。最后的參數是一個Func<string,Task>類型的委托，用來處理接收到的響應內容（字符串），在這里我們依然是將格式化的響應內容直接打印在控制臺上。
程序啟動后控制臺上將出現如上圖所示的輸出，客戶端/服務端接收內容的交錯輸出體現了我們希望的“雙向流式”消息交換模式。我們將在后續介紹HanleRequestAsync和SendStreamRequestAsync方法的實現邏輯。 二、[服務端]流式請求/響應的讀寫HanleRequestAsync方法定義如下。如代碼片段所示，我們利用請求的BodyReader和響應的BodyWriter來對請求和響應內容進行讀寫，它們的類型分別是PipeReader和PipeWriter。在一個循環中，在利用BodyReader將請求緩沖區內容讀取出來后，我們將得到的ReadOnlySequence<byte>對象作為參數調用輔助方法TryReadMessage讀取單條請求消息，并調用handler參數表示的處理器進行處理。當請求內容接收完畢后，循環終止。 static async Task HandleRequestAsync(HttpContext httpContext, Func<string, PipeWriter, Task> handler)
{
var reader = httpContext.Request.BodyReader;
var writer = httpContext.Response.BodyWriter;
while (true)
{
var result = await reader.ReadAsync();
var buffer = result.Buffer;
while (TryReadMessage(ref buffer, out var message))
{
await handler(message, writer);
}
reader.AdvanceTo(buffer.Start, buffer.End);
if (result.IsCompleted)
{
break;
}
}
}由于客戶端發送的單條字符串消息長度不限，為了精準地將其讀出來，我們需要在輸出編碼后的消息內容前添加4個字節的整數來表示消息的長度。所以在如下所示的TryReadMessage方法中，我們會先將字節長度讀取出來，再據此將消息自身內容讀取出來，最終通過解碼得到消息字符串。 static bool TryReadMessage(ref ReadOnlySequence<byte> buffer, [NotNullWhen(true)]out string? message)
{
var reader = new SequenceReader<byte>(buffer);
if (!reader.TryReadLittleEndian(out int length))
{
message = default;
return false;
}
message = Encoding.UTF8.GetString(buffer.Slice(4, length));
buffer = buffer.Slice(length + 4);
return true;
}響應消息的寫入是通過如下針對PipeWriter的WriteStringAsync擴展方法實現的，這里的PipeWriter就是響應的BodyWriter，針對“Length + Payload“的消息寫入也體現在這里。 public static class Extensions
{
public static ValueTask<FlushResult> WriteStringAsync(this PipeWriter writer, string content)
{
var length = Encoding.UTF8.GetByteCount(content);
var span = writer.GetSpan(4 + length);
BitConverter.TryWriteBytes(span, length);
Encoding.UTF8.GetBytes(content, span.Slice(4));
writer.Advance(4 + length);
return writer.FlushAsync();
}
}三、[客戶端]流式響應/請求的讀寫客戶端利用HttpClient發送請求。針對HttpClient的請求通過一個HttpRequestMessage對象表示，其主體內容體現為一個HttpContent。流式請求的發送是通過如下這個StreamContent類型實現的，它派生于HttpContent。我們重寫了SerializeToStreamAsync方法，利用自定義的StreamContentWriter將內容寫入請求輸出流。 public class StreamContent(StreamContentWriter writer) : HttpContent
{
private readonly StreamContentWriter _writer = writer;
protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context)
=> _writer.SetOutputStream(stream).WaitAsync();
protected override bool TryComputeLength(out long length) => (length = -1) != -1;
}
public class StreamContentWriter
{
private readonly TaskCompletionSource<Stream> _streamSetSource = new();
private readonly TaskCompletionSource _streamEndSource = new();
public StreamContentWriter SetOutputStream(Stream outputStream)
{
_streamSetSource.SetResult(outputStream);
return this;
}
public async Task WriteAsync(string content)
{
var stream = await _streamSetSource.Task;
await PipeWriter.Create(stream).WriteStringAsync(content);
}
public void Complete() => _streamEndSource.SetResult();
public Task WaitAsync() => _streamEndSource.Task;
}StreamContentWriter提供了四個方法，SetOutputStream方法用來設置請求輸出流，上面重寫的SerializeToStreamAsync調用了此方法。單條字符串消息的寫入實現在WriteAsync方法中，它最終調用的依然是上面提供的WriteStringAsync擴展方法。整個流式請求的過程通過一個TaskCompletionSource對象提供的Task來表示，當客戶端完成所有輸出后，會調用Complete方法，該方法進一步調用這個TaskCompletionSource對象的SetResult方法。由于WaitAsync方法返回TaskCompletionSource對象提供的Task，SerializeToStreamAsync方法會調用此方法等待”客戶端輸出流“的終結。 如下的代碼片段體現了SendStreamRequestAsync方法的實現。在這里我們創建了一個表示流式請求的HttpRequestMessage對象，我們將協議版本設置為HTTP2，作為主體內容的HttpContent正式根據StreamContentWriter對象創建的StreamContent對象。 static async Task SendStreamRequestAsync(string url,string[] lines, Func<string, Task> handler)
{
using var httpClient = new HttpClient();
var writer = new StreamContentWriter();
var request = new HttpRequestMessage(HttpMethod.Post, url)
{
Version = HttpVersion.Version20,
VersionPolicy = HttpVersionPolicy.RequestVersionExact,
Content = new StreamingWeb.StreamContent(writer)
};
var task = httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead);
_ = Task.Run(async () =>
{
var response = await task;
var reader = PipeReader.Create(await response.Content.ReadAsStreamAsync());
while (true)
{
var result = await reader.ReadAsync();
var buffer = result.Buffer;
while (TryReadMessage(ref buffer, out var message))
{
await handler(message);
}
reader.AdvanceTo(buffer.Start, buffer.End);
if (result.IsCompleted)
{
break;
}
}
});
foreach (string line in lines)
{
await writer.WriteAsync($"{line} ({DateTimeOffset.UtcNow})");
await Task.Delay(1000);
}
writer.Complete();
}我們將這個HttpRequestMessage作為請求利用HttpClient發送出去，實際上發送的內容最終是通過調用StreamContentWriter對象的WriteAsync方法輸出的，我們每隔1秒發送一條消息。HttpClient將請求發出去之后會得到一個通過HttpResponseMessage對象表示的響應，在一個異步執行的Task中，我們根據響應流創建一個PipeReader對象，并在一個循環中調用上面定義的TryReadMessage方法逐條讀取接收到的單條消息進行處理。 本文作者：Artech，轉自https://www.cnblogs.com/artech/p/18021662/streaming_messaging 該文章在 2024/2/24 15:32:33 編輯過 |
關鍵字查詢
相關文章
正在查詢... 
|
400 186 1886
