引入

項目創建

mkdir RpcStreaming
cd RpcStreaming
dotnet new console
dotnet add package Grpc // 添加 gRPC 包
dotnet add package Grpc.Tools // 添加 gRPC 工具包
dotnet add package Google.Protobuf // 添加 Protobuf 支持

然後為了支持 protobuf 語言，我們需要修改項目配置文件，在項目中引入 .proto 文件以便生成對應的代碼。

<Project Sdk="Microsoft.NET.Sdk">
  ...
  <PropertyGroup>
    ...
    <LangVersion>latest</LangVersion>
    ...
  </PropertyGroup>
  
  <ItemGroup>
    ...
    <Protobuf Include="**/*.proto" />
    ...
  </ItemGroup>
  ...
</Project>

這裡我們使用了 wildcard 語法匹配了項目內的全部 proto 文件用於生成對應的代碼。

編寫 Proto 文件

 1 synatx = "proto3";
 2 service RpcStreamingService {
 3   rpc GetStreamContent (StreamRequest) returns (stream StreamContent) {}
 4 }
 5 message StreamRequest {
 6   string fileName = 1;
 7 }
 8 message StreamContent {
 9   bytes content = 1;
10 }

做 RPC 請求時，我們向 RPC 伺服器發送一個 StreamRequest 的 message，其中包含了文件路徑；為了讓伺服器以流式傳輸數據，我們在 returns 內加一個 “stream”。

編寫 Server 端代碼

public virtual global::System.Threading.Tasks.Task GetStreamContent(global::StreamRequest request, grpc::IServerStreamWriter<global::StreamContent> responseStream, grpc::ServerCallContext context)
{
    throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, ""));
}

這樣就簡單了，我們新建一個類 RpcServiceImpl，繼承 RpcStreamingService.RpcStreamingServiceBase，然後實現對應的方法即可。

 1 using System;
 2 using System.IO;
 3 using System.Threading.Tasks;
 4 using Google.Protobuf;
 5 using Grpc.Core;
 6 namespace RpcStreaming
 7 {
 8     public class RpcStreamingServiceImpl : RpcStreamingService.RpcStreamingServiceBase
 9     {
10         public override Task GetStreamContent(StreamRequest request, IServerStreamWriter<StreamContent> response, ServerCallContext context)
11         {
12             return Task.Run(async () =>
13             {
14                 using (var fs = File.Open(request.FileName, FileMode.Open)) // 從 request 中讀取文件名並打開文件流
15           {
16                     var remainingLength = fs.Length; // 剩餘長度
17               var buff = new byte[1048576]; // 緩衝區，這裡我們設置為 1 Mb
18               while (remainingLength > 0) // 若未讀完則繼續讀取
19               {
20                         var len = await fs.ReadAsync(buff); // 非同步從文件中讀取數據到緩衝區中
21                   remainingLength -= len; // 剩餘長度減去剛纔實際讀取的長度
22 
23                   // 向流中寫入我們剛剛讀取的數據
24                   await response.WriteAsync(new StreamContent
25                         {
26                             Content = ByteString.CopyFrom(buff, 0, len)
27                         });
28                     }
29                 }
30             });
31         }
32     }
33 }

啟動 RPC Server

1 using Google.Protobuf;
2 using Grpc.Core;

然後我們在 Main 函數中構建並啟動 RPC Server，監聽 localhost:23333

1 new Server
2 {
3     Services = { RpcStreamingService.BindService(new RpcStreamingServiceImpl()) }, // 綁定我們的實現
4     Ports = { new ServerPort("localhost", 23333, ServerCredentials.Insecure) }
5 }.Start();
6 Console.ReadKey();

這樣服務端就構建完成了。

編寫客戶端調用 RPC API

1 // 原來的 Main 函數
2 static void Main(string[] args) { ... }
3 // 改寫後的 Main 函數
4 static async Task Main(string[] args) { ... }

另外，還需要：

1 using System;
2 using System.IO;
3 using System.Threading.Tasks;
4 using Google.Protobuf;
5 using Grpc.Core;

然後我們在 Main 函數中添加調用代碼：

 1 var channel = new Channel("localhost:23333", ChannelCredentials.Insecure); // 建立到 localhost:23333 的 channel
 2 var client = new RpcStreamingService.RpcStreamingServiceClient(channel); // 建立 client
 3 // 調用 RPC API
 4 var result = client.GetStreamContent(new StreamRequest { FileName = "你想獲取的文件路徑" });
 5 var iter = result.ResponseStream; // 拿到響應流
 6 using (var fs = new FileStream("寫獲取的數據的文件路徑", FileMode.Create)) // 新建一個文件流用於存放我們獲取到數據
 7 {
 8     while (await iter.MoveNext()) // 迭代
 9     {
10         iter.Current.Content.WriteTo(fs); // 將數據寫入到文件流中
11     }
12 }

測試

dotnet run

會發現，我們想要獲取的文件的數據被不斷地寫到我們指定的文件中，每次 1 Mb。在我的電腦上測試，內網的環境下傳輸速度大概 80~90 Mb/s，幾乎跑滿了我的千兆網卡，速度非常理想。