(精華)2020年10月28日 Grpc Grpc雙向流調(diào)用

一.前言

gRPC 支持雙向流調(diào)用，支持實(shí)時(shí)推送消息，這也是 gRPC的一大特點(diǎn)，且 gRPC 在對(duì)雙向流的控制支持上也是非常強(qiáng)大的。

二. 什么是 gRPC 流

gRPC 有四種服務(wù)類型，分別是：簡單 RPC（Unary RPC）、服務(wù)端流式 RPC （Server streaming RPC）、客戶端流式 RPC （Client streaming RPC）、雙向流式 RPC（Bi-directional streaming RPC）。它們主要有以下特點(diǎn)：

三.為什么 gRPC 支持流

gRPC 通信是基于 HTTP/2 實(shí)現(xiàn)的，它的雙向流映射到 HTTP/2 流。HTTP/2 具有流的概念，流是為了實(shí)現(xiàn)HTTP/2的多路復(fù)用。流是服務(wù)器和客戶端在HTTP/2連接內(nèi)用于交換幀數(shù)據(jù)的獨(dú)立雙向序列，邏輯上可看做一個(gè)較為完整的交互處理單元，即表達(dá)一次完整的資源請(qǐng)求、響應(yīng)數(shù)據(jù)交換流程；一個(gè)業(yè)務(wù)處理單元，在一個(gè)流內(nèi)進(jìn)行處理完畢，這個(gè)流生命周期完結(jié)。

特點(diǎn)如下：

一個(gè)HTTP/2連接可同時(shí)保持多個(gè)打開的流，任一端點(diǎn)交換幀

流可被客戶端或服務(wù)器單獨(dú)或共享創(chuàng)建和使用

流可被任一端關(guān)閉

在流內(nèi)發(fā)送和接收數(shù)據(jù)都要按照順序

流的標(biāo)識(shí)符自然數(shù)表示，1~2^31-1區(qū)間，有創(chuàng)建流的終端分配

流與流之間邏輯上是并行、獨(dú)立存在

四.gRPC中使用雙向流調(diào)用

我們?cè)谇拔闹芯帉懙腞PC屬于簡單RPC，沒有包含流調(diào)用，下面直接講一下雙向流，根據(jù)第二小節(jié)列舉的四種服務(wù)類型，如果我們掌握了簡單RPC和雙向流RPC，那么服務(wù)端流式 RPC和客戶端流式 RPC自然也就沒有問題了。

這里我們繼續(xù)使用前文的代碼，要實(shí)現(xiàn)的目標(biāo)是一次給多個(gè)貓洗澡。

① 首先在 LuCat.proto 定義兩個(gè)rpc，一個(gè) Count 用于統(tǒng)計(jì)貓的數(shù)量，一個(gè) 雙向流 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)計(jì)貓數(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ù)的實(shí)現(xiàn)

private readonly ILogger _logger; private static readonly List Cats=new List(){"英短銀漸層","英短金漸層","美短","藍(lán)貓","貍花貓","橘貓"}; private static readonly Random Rand=new Random(DateTime.Now.Millisecond); public LuCatService(ILogger logger) { _logger = logger; } public override async Task BathTheCat(IAsyncStreamReader requestStream, IServerStreamWriter responseStream, ServerCallContext context) { var bathQueue=new Queue(); while (await requestStream.MoveNext()) { //將要洗澡的貓加入隊(duì)列 bathQueue.Enqueue(requestStream.Current.Id); _logger.LogInformation($"Cat {requestStream.Current.Id} Enqueue."); } //遍歷隊(duì)列開始洗澡 while (bathQueue.TryDequeue(out var catId)) { await responseStream.WriteAsync(new BathTheCatResp() { Message = $"鏟屎的成功給一只{Cats[catId]}洗了澡！" }); await Task.Delay(500);//此處主要是為了方便客戶端能看出流調(diào)用的效果 } } public override Task Count(Empty request, ServerCallContext context) { return Task.FromResult(new CountCatResult() { Count = Cats.Count }); }

BathTheCat 方法會(huì)接收多個(gè)客戶端發(fā)來的CatId，然后將他們加入隊(duì)列中，等客戶端發(fā)送完成后開始依次洗澡并返回給客戶端。

③ 客戶端實(shí)現(xiàn)

隨機(jī)發(fā)送10個(gè)貓Id給服務(wù)端，然后接收10個(gè)洗澡結(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個(gè)貓洗澡 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個(gè)需要洗澡的貓id"); Console.WriteLine("接收洗澡結(jié)果："); //開始接收響應(yīng) await bathCatRespTask; Console.WriteLine("洗澡完畢");

④ 運(yùn)行

可以看到雙向流調(diào)用成功，客戶端發(fā)送了10個(gè)貓洗澡請(qǐng)求對(duì)象，服務(wù)端返回了10個(gè)貓洗澡結(jié)果對(duì)象。且是實(shí)時(shí)推送的，這就是 gRPC 的雙向流調(diào)用。

這里大家可以自行改進(jìn)來演變成客戶端流式或者服務(wù)端流式調(diào)用。客戶端發(fā)送一個(gè)貓Id列表，然后服務(wù)端返回每個(gè)貓洗澡結(jié)果，這就是服務(wù)端流式調(diào)用。客戶端依次發(fā)送貓Id，然后服務(wù)端一次性返回所有貓的洗澡結(jié)果（給所有貓洗澡看做是一個(gè)業(yè)務(wù)，返回這個(gè)業(yè)務(wù)的結(jié)果），就是客戶端流式調(diào)用。這里我就不再演示了。

五.流控制

gRPC 的流式調(diào)用支持對(duì)流進(jìn)行主動(dòng)取消的控制，進(jìn)而可以衍生出流超時(shí)限制等控制。

在流式調(diào)用是，可以傳一個(gè) CancellationToken 參數(shù)，它就是我們用來對(duì)流進(jìn)行取消控制的：

改造一下我們?cè)诘谒男」?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ù)端

//遍歷隊(duì)列開始洗澡 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個(gè)貓的洗澡返回結(jié)果，流就已經(jīng)被取消了，這就是 gRPC 的流控制。

六.結(jié)束

這里流式調(diào)用可以實(shí)現(xiàn)實(shí)時(shí)推送，服務(wù)端到客戶端或者客戶端到服務(wù)端短實(shí)時(shí)推送消息，但是這個(gè)和傳統(tǒng)意義上的長連接主動(dòng)推送、廣播消息不一樣，不管你是服務(wù)端流式、客戶端流式還是雙向流式，必須要由客戶端進(jìn)行發(fā)起，通過客戶端請(qǐng)求來建立流通信。

HTTP RPC

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請(qǐng)聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請(qǐng)聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。