面试题：C# 中 gRPC 服务端如何实现流处理

Question

在基于 C# 的 gRPC 微服务开发中，假设你要实现一个文件上传功能，需要服务端以流的形式接收客户端发送的文件数据并进行处理，描述实现这一功能的主要步骤以及相关代码示例。

Answer 1

1. 定义服务契约

首先在 .proto 文件中定义服务接口和消息类型。

syntax = "proto3";

package fileupload;

service FileUploadService {
  rpc Upload(stream UploadRequest) returns (UploadResponse);
}

message UploadRequest {
  bytes data = 1;
}

message UploadResponse {
  string message = 1;
}

2. 生成 C# 代码

使用 protoc 工具根据 .proto 文件生成 C# 代码。假设 fileupload.proto 是上述定义的文件，命令如下：

protoc -I. --csharp_out=. fileupload.proto

3. 实现服务端

3.1 创建 gRPC 服务

在 C# 项目中创建一个类继承自生成的 FileUploadServiceBase 类，并实现 Upload 方法。

using Grpc.Core;
using System.IO;
using System.Threading.Tasks;

public class FileUploadServiceImpl : FileUploadService.FileUploadServiceBase
{
    public override async Task<UploadResponse> Upload(IAsyncStreamReader<UploadRequest> requestStream, ServerCallContext context)
    {
        using (var fileStream = new FileStream("uploadedFile", FileMode.Create, FileAccess.Write))
        {
            while (await requestStream.MoveNext())
            {
                var buffer = requestStream.Current.Data.ToByteArray();
                await fileStream.WriteAsync(buffer, 0, buffer.Length);
            }
        }
        return new UploadResponse { Message = "File uploaded successfully" };
    }
}

3.2 配置 gRPC 服务

在 Startup.cs 文件中配置 gRPC 服务。

using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddGrpc();
    }

    public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
    {
        if (env.IsDevelopment())
        {
            app.UseDeveloperExceptionPage();
        }

        app.UseRouting();

        app.UseEndpoints(endpoints =>
        {
            endpoints.MapGrpcService<FileUploadServiceImpl>();
        });
    }
}

4. 实现客户端

4.1 创建 gRPC 客户端

using Grpc.Net.Client;
using System;
using System.IO;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        using var channel = GrpcChannel.ForAddress("http://localhost:5000");
        var client = new FileUploadService.FileUploadServiceClient(channel);

        using var call = client.Upload();

        var filePath = "localFileToUpload";
        using (var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read))
        {
            var buffer = new byte[1024];
            int read;
            while ((read = fileStream.Read(buffer, 0, buffer.Length)) > 0)
            {
                var request = new UploadRequest { Data = Google.Protobuf.ByteString.CopyFrom(buffer, 0, read) };
                await call.RequestStream.WriteAsync(request);
            }
        }

        await call.RequestStream.CompleteAsync();
        var response = await call;
        Console.WriteLine(response.Message);
    }
}