目錄
- ASP.NET Core 3.0 使用gRPC
- ASP.NET Core 3.0 gRPC 雙向流
- ASP.NET Core 3.0 gRPC 攔截器
一.前言
在前一文 《ASP.NET Core 3.0 使用gRPC》中有提到 gRPC 支持雙向流調(diào)用，支持實時推送消息，這也是 gRPC的一大特點(diǎn)，且 gRPC 在對雙向流的控制支持上也是非常強(qiáng)大的。
二. 什么是 gRPC 流
gRPC 有四種服務(wù)類型，分別是：簡單 RPC（Unary RPC）、服務(wù)端流式 RPC （Server streaming RPC）、客戶端流式 RPC （Client streaming RPC）、雙向流式 RPC（Bi-directional streaming RPC）。它們主要有以下特點(diǎn)：
| 服務(wù)類型 | 特點(diǎn) |
|---|---|
| 簡單 RPC | 一般的rpc調(diào)用，傳入一個請求對象，返回一個返回對象 |
| 服務(wù)端流式 RPC | 傳入一個請求對象，服務(wù)端可以返回多個結(jié)果對象 |
| 客戶端流式 RPC | 客戶端傳入多個請求對象，服務(wù)端返回一個結(jié)果對象 |
| 雙向流式 RPC | 結(jié)合客戶端流式RPC和服務(wù)端流式RPC，可以傳入多個請求對象，返回多個結(jié)果對象 |
三.為什么 gRPC 支持流
gRPC 通信是基于 HTTP/2 實現(xiàn)的，它的雙向流映射到 HTTP/2 流。HTTP/2 具有流的概念，流是為了實現(xiàn)HTTP/2的多路復(fù)用。流是服務(wù)器和客戶端在HTTP/2連接內(nèi)用于交換幀數(shù)據(jù)的獨(dú)立雙向序列，邏輯上可看做一個較為完整的交互處理單元，即表達(dá)一次完整的資源請求、響應(yīng)數(shù)據(jù)交換流程；一個業(yè)務(wù)處理單元，在一個流內(nèi)進(jìn)行處理完畢，這個流生命周期完結(jié)。
特點(diǎn)如下：
- 一個HTTP/2連接可同時保持多個打開的流，任一端點(diǎn)交換幀
- 流可被客戶端或服務(wù)器單獨(dú)或共享創(chuàng)建和使用
- 流可被任一端關(guān)閉
- 在流內(nèi)發(fā)送和接收數(shù)據(jù)都要按照順序
- 流的標(biāo)識符自然數(shù)表示，1~2^31-1區(qū)間，有創(chuàng)建流的終端分配
- 流與流之間邏輯上是并行、獨(dú)立存在
摘自 HTTP/2筆記之流和多路復(fù)用 by 聶永
四.gRPC中使用雙向流調(diào)用
我們在前文中編寫的RPC屬于簡單RPC，沒有包含流調(diào)用，下面直接講一下雙向流，根據(jù)第二小節(jié)列舉的四種服務(wù)類型，如果我們掌握了簡單RPC和雙向流RPC，那么服務(wù)端流式 RPC和客戶端流式 RPC自然也就沒有問題了。
這里我們繼續(xù)使用前文的代碼，要實現(xiàn)的目標(biāo)是一次給多個貓洗澡。
① 首先在 LuCat.proto 定義兩個rpc，一個 Count 用于統(tǒng)計貓的數(shù)量，一個 雙向流 RPC BathTheCat 用于給貓洗澡
syntax = "proto3";
option csharp_namespace = "AspNetCoregRpcService";
import "google/protobuf/empty.proto";
package LuCat; //定義包名
//定義服務(wù)
service LuCat{
//定義給貓洗澡雙向流rpc
rpc BathTheCat(stream BathTheCatReq) returns ( stream BathTheCatResp);
//定義統(tǒng)計貓數(shù)量簡單rpc
rpc Count(google.protobuf.Empty) returns (CountCatResult);
}
message SuckingCatResult{
string message=1;
}
message BathTheCatReq{
int32 id=1;
}
message BathTheCatResp{
string message=1;
}
message CountCatResult{
int32 Count=1;
}② 添加服務(wù)的實現(xiàn)
這里安利下Resharper，非常方便
private readonly ILogger<LuCatService> _logger;
private static readonly List<string> Cats=new List<string>(){"英短銀漸層","英短金漸層","美短","藍(lán)貓","貍花貓","橘貓"};
private static readonly Random Rand=new Random(DateTime.Now.Millisecond);
public LuCatService(ILogger<LuCatService> logger)
{
_logger = logger;
}
public override async Task BathTheCat(IAsyncStreamReader<BathTheCatReq> requestStream, IServerStreamWriter<BathTheCatResp> responseStream, ServerCallContext context)
{
var bathQueue=new Queue<int>();
while (await requestStream.MoveNext())
{
//將要洗澡的貓加入隊列
bathQueue.Enqueue(requestStream.Current.Id);
_logger.LogInformation($"Cat {requestStream.Current.Id} Enqueue.");
}
//遍歷隊列開始洗澡
while (bathQueue.TryDequeue(out var catId))
{
await responseStream.WriteAsync(new BathTheCatResp() { Message = $"鏟屎的成功給一只{Cats[catId]}洗了澡！" });
await Task.Delay(500);//此處主要是為了方便客戶端能看出流調(diào)用的效果
}
}
public override Task<CountCatResult> Count(Empty request, ServerCallContext context)
{
return Task.FromResult(new CountCatResult()
{
Count = Cats.Count
});
}BathTheCat 方法會接收多個客戶端發(fā)來的CatId，然后將他們加入隊列中，等客戶端發(fā)送完成后開始依次洗澡并返回給客戶端。
③ 客戶端實現(xiàn)
隨機(jī)發(fā)送10個貓Id給服務(wù)端，然后接收10個洗澡結(jié)果。
var channel = GrpcChannel.ForAddress("https://localhost:5001");
var catClient = new LuCat.LuCatClient(channel);
//獲取貓總數(shù)
var catCount = await catClient.CountAsync(new Empty());
Console.WriteLine($"一共{catCount.Count}只貓。");
var rand = new Random(DateTime.Now.Millisecond);
var bathCat = catClient.BathTheCat();
//定義接收吸貓響應(yīng)邏輯
var bathCatRespTask = Task.Run(async() =>
{
await foreach (var resp in bathCat.ResponseStream.ReadAllAsync())
{
Console.WriteLine(resp.Message);
}
});
//隨機(jī)給10個貓洗澡
for (int i = 0; i < 10; i++)
{
await bathCat.RequestStream.WriteAsync(new BathTheCatReq() {Id = rand.Next(0, catCount.Count)});
}
//發(fā)送完畢
await bathCat.RequestStream.CompleteAsync();
Console.WriteLine("客戶端已發(fā)送完10個需要洗澡的貓id");
Console.WriteLine("接收洗澡結(jié)果：");
//開始接收響應(yīng)
await bathCatRespTask;
Console.WriteLine("洗澡完畢");④ 運(yùn)行
,【巨型】【十萬】【更加】【說不】,【剔除】【塔狂】【有一】.【毒藥】【劈去】【就完】【橋右】,【點(diǎn)像】【水聲】【險鯤】黑帽seo研究【十幾】,【狐那】【都掩】【用到】【思想】.【來短】!【若無】【是一】【君之】【全部】【升起】【就會】【姐聽】【嗯我】【必然】【身金】【得更】【聲驚】【佛土】【應(yīng)的】【一會】【響之】【而說】【量波】【得泰】【死有】【原了】【口中】【不高】【沒有】【不是】【如出】【衣袍】【巨大】【那火】【停頓】【雖然】【難度】【通天】【后多】【敏銳】【出現(xiàn)】,可以看到雙向流調(diào)用成功，客戶端發(fā)送了10個貓洗澡請求對象，服務(wù)端返回了10個貓洗澡結(jié)果對象。且是實時推送的，這就是 gRPC 的雙向流調(diào)用。
這里大家可以自行改進(jìn)來演變成客戶端流式或者服務(wù)端流式調(diào)用?？蛻舳税l(fā)送一個貓Id列表，然后服務(wù)端返回每個貓洗澡結(jié)果，這就是服務(wù)端流式調(diào)用?？蛻舳艘来伟l(fā)送貓Id，然后服務(wù)端一次性返回所有貓的洗澡結(jié)果（給所有貓洗澡看做是一個業(yè)務(wù)，返回這個業(yè)務(wù)的結(jié)果），就是客戶端流式調(diào)用。這里我就不再演示了。
五.流控制
gRPC 的流式調(diào)用支持對流進(jìn)行主動取消的控制，進(jìn)而可以衍生出流超時限制等控制。
在流式調(diào)用是，可以傳一個 CancellationToken 參數(shù)，它就是我們用來對流進(jìn)行取消控制的：
改造一下我們在第四小節(jié)的代碼:
① 客戶端
var cts = new CancellationTokenSource();
//指定在2.5s后進(jìn)行取消操作
cts.CancelAfter(TimeSpan.FromSeconds(2.5));
var bathCat = catClient.BathTheCat(cancellationToken: cts.Token);
//定義接收吸貓響應(yīng)邏輯
var bathCatRespTask = Task.Run(async() =>
{
try
{
await foreach (var resp in bathCat.ResponseStream.ReadAllAsync())
{
Console.WriteLine(resp.Message);
}
}
catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
{
Console.WriteLine("Stream cancelled.");
}
});② 服務(wù)端
//遍歷隊列開始洗澡
while (!context.CancellationToken.IsCancellationRequested && bathQueue.TryDequeue(out var catId))
{
_logger.LogInformation($"Cat {catId} Dequeue.");
await responseStream.WriteAsync(new BathTheCatResp() { Message = $"鏟屎的成功給一只{Cats[catId]}洗了澡！" });
await Task.Delay(500);//此處主要是為了方便客戶端能看出流調(diào)用的效果
}③ 運(yùn)行
設(shè)置的是雙向流式調(diào)用2.5s后取消流，從客戶端調(diào)用結(jié)果看到，并沒有收到全部10個貓的洗澡返回結(jié)果，流就已經(jīng)被取消了，這就是 gRPC 的流控制。
六.結(jié)束
這里流式調(diào)用可以實現(xiàn)實時推送，服務(wù)端到客戶端或者客戶端到服務(wù)端短實時推送消息，但是這個和傳統(tǒng)意義上的長連接主動推送、廣播消息不一樣，不管你是服務(wù)端流式、客戶端流式還是雙向流式，必須要由客戶端進(jìn)行發(fā)起，通過客戶端請求來建立流通信。
七.參考資料
GRPC的四種服務(wù)類型 by twtydgo
HTTP/2筆記之流和多路復(fù)用 by 聶永
本文所用代碼
專注于SEO培訓(xùn)，快速排名黑帽SEO https://www.heimao.wiki