使用双向（双工）流式传输的 C#/.NET gRPC 服务

• 从 Github 下载完整源代码

引言

gRPC 优于 REST API 的主要优势之一是 gRPC 除了传统的请求-响应方式外，还支持流式传输方法。gRPC 中有 3 种流式传输支持，分别是客户端流式传输、服务器流式传输和双向（双工）流式传输。在本文中，我们将讨论双向（双工）流式传输以及如何在 C#/.NET 中实现它。

什么是双工流式传输

在双工流式传输场景中，客户端和服务器都通过单独的读取和写入流相互发送一系列消息。调用由客户端发起到服务器，之后流将可用。流彼此独立，因此客户端和服务器可以根据其自身应用程序的需求读取和写入流。例如，服务器可以等待来自客户端的所有消息，然后再发回响应，或者它可以立即回复并与客户端进行“乒乓”式的、类似聊天的通信。在每个流中，消息的顺序是有保证的。

下面显示了双工流式传输方法的定义。请注意请求和响应中都使用了关键字 stream。

rpc ChatNotification(stream NotificationsRequest) returns (stream NotificationsResponse);

实现（服务器）

要创建一个 gRPC 项目，启动 Visual Studio 并运行项目模板 ASP.NET Core gRPC Service。默认情况下，dotnet 会创建一个 greet.proto 文件，其中包含生成的 GreeterService 的服务定义。让我们编辑这个文件，并将其内容替换为下面的 proto 文件。此外，最好重命名服务以及 proto 文件。在此示例代码中，让我们将服务从 Greeter 重命名为 Notifier，并将 greet.proto 重命名为 notify.proto。

引用

在此示例实现中，我们将进行“乒乓”式的通信，其中服务器响应客户端发出的每个请求。

syntax = "proto3";
import "google/protobuf/timestamp.proto";
option csharp_namespace = "DuplexStreaming";

package notify;

service Notifer {
   rpc ChatNotification(stream NotificationsRequest) returns (stream NotificationsResponse);  
}

message NotificationsRequest {
  string message = 1;
  string to = 2;
  string from = 3;
}

message NotificationsResponse{  
  string message = 1;
  google.protobuf.Timestamp receivedAt = 3;
}

构建项目/解决方案以确保一切正确。现在，打开 *NotifierService.cs* 并添加 ChatNotifications 方法的实现。

using System.ComponentModel;
using System.Diagnostics.Metrics;

using DuplexStreaming;

using Google.Protobuf.WellKnownTypes;

using Grpc.Core;

namespace DuplexStreaming.Services;
public class NotifierService : Notifier.NotifierBase {
    private readonly ILogger<NotifierService> _logger;
    public NotifierService(ILogger<NotifierService> logger) {
        _logger = logger;
    }

    public override async Task ChatNotification(IAsyncStreamReader<NotificationsRequest> 
      requestStream, IServerStreamWriter<NotificationsResponse> responseStream, 
      ServerCallContext context) {

        while (await requestStream.MoveNext()) {
            var request = requestStream.Current;
          
            var now = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow);
            var reply = new NotificationsResponse() {
                Message = $"Hi {request.From}!, 
                You have sent the message \"{request.Message}\" to {request.To}",
                ReceivedAt = now
            };
         
            await responseStream.WriteAsync(reply);          
        }
    }
}

引用

注意，该方法的第一个参数是 IAsyncStreamReader<NotificationRequest>，第二个参数是 IServerStreamWriter<NotificationResponse>

• IAsyncStreamReader<T> requestStream
  – 客户端应用程序写入此流以向服务器发送消息
• IServerStreamWriter<T> responseStream
  – 服务器写入此流以向客户端发送消息。在我们的示例实现中，服务器会为从客户端收到的每条消息发回一个响应（就像一个聊天应用程序）

查看上面的代码，我们在 requestStream 上调用 MoveNext()，然后通过访问 Current 属性从客户端获取最新消息。然后，我们使用请求中的值通过调用 responseStream 的 WriteAsync 方法立即发回响应。

实现（客户端）

using DuplexStreaming;
using Grpc.Net.Client;

// The port number must match the port of the gRPC server.
using var channel = GrpcChannel.ForAddress("https://:5295");
var client = new Notifier.NotifierClient(channel);
using var call = client.ChatNotification();

var responseReaderTask = Task.Run(async Task () =>
{
    while (await call.ResponseStream.MoveNext(CancellationToken.None))
    {
        var note = call.ResponseStream.Current;
        Console.WriteLine($"{note.Message}, received at {note.ReceivedAt}");
    }
});

foreach (var msg in new[] {"Tom", "Jones"})
{
    var request = new NotificationsRequest() 
                  { Message = $"Hello {msg}", From = "Mom", To = msg };
    await call.RequestStream.WriteAsync(request);
}

await call.RequestStream.CompleteAsync();
await responseReaderTask;

Console.WriteLine("Press any key to exit...");
Console.ReadKey();

在客户端，我们需要通过调用 client.ChatNotification() 来发起调用，以获得请求和响应流。之后，我们设置一个任务来读取服务器响应，然后再将消息写入请求流。

现在，运行服务器，然后运行控制台应用程序客户端。您应该在控制台客户端上看到类似这样的内容...

PS C:\Users\Erik\Source\Repos\grpctutorials\DuplexStreaming\source\DuplexStreamingClient\bin\debug\net8.0> .\DuplexStreamingClient.exe      
Hi Mom!, You have sent the message "Hello Tom" to Tom, 
received at "2024-01-25T10:07:32.183720200Z"
Hi Mom!, You have sent the message "Hello Jones" to Jones, 
received at "2024-01-25T10:07:32.183947200Z"
Press any key to exit...

使用 FintX 测试

您也可以在不需要编写客户端应用程序的情况下测试该服务。让我们启动 FintX (https://github.com/namigop/FintX) 来验证该服务是否正常工作。FintX 是一个开源、原生和跨平台的 gRPC 客户端。我在 Windows 上，所以我安装了 Windows 包 (MacOS 和 Linux 下载也可用)

单击加号图标以添加客户端。在 http 地址中输入的值应与正在运行的服务匹配

单击确定，然后双击 ChatNotifications 方法以在新选项卡中打开它。通过单击绿色运行按钮启动调用 - 之后将一些数据写入请求流。

在测试结束时，您应该看到类似下面屏幕截图的内容

下面是一个简短的视频，展示了如何使用 FintX 测试双工 gRPC 服务的步骤。

祝您编码愉快！