Tạo gRPC services và methods

Bài viết này giải thích cách tạo gRPC services (dịch vụ gRPC) và methods (phương thức) trong C#. Các chủ đề bao gồm:

Cách định nghĩa services và methods trong tệp .proto.
Mã được tạo bằng công cụ gRPC C#.
Triển khai gRPC services và methods.

Tạo gRPC services mới

gRPC services với C# đã giới thiệu cách tiếp cận contract-first (hợp đồng trước) của gRPC để phát triển API. Services và messages được định nghĩa trong tệp .proto. Công cụ C# sau đó tạo mã từ tệp .proto. Đối với server-side assets (tài sản phía server), một abstract base type (kiểu cơ sở trừu tượng) được tạo cho mỗi service, cùng với các lớp cho bất kỳ messages nào.

Tệp .proto sau:

Định nghĩa một Greeter service.
Greeter service định nghĩa một SayHello call.
SayHello gửi một HelloRequest message và nhận một HelloReply message

protobuf

syntax = "proto3";

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply);
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

Công cụ C# tạo ra kiểu cơ sở GreeterBase của C#:

csharp

public abstract partial class GreeterBase
{
    public virtual Task<HelloReply> SayHello(HelloRequest request, ServerCallContext context)
    {
        throw new RpcException(new Status(StatusCode.Unimplemented, ""));
    }
}

public class HelloRequest
{
    public string Name { get; set; }
}

public class HelloReply
{
    public string Message { get; set; }
}

Theo mặc định, GreeterBase được tạo không làm gì cả. Virtual method SayHello của nó sẽ trả về lỗi UNIMPLEMENTED cho bất kỳ client nào gọi nó. Để service hoạt động hữu ích, ứng dụng phải tạo một concrete implementation (triển khai cụ thể) của GreeterBase:

csharp

public class GreeterService : GreeterBase
{
    public override Task<HelloReply> SayHello(HelloRequest request, ServerCallContext context)
    {
        return Task.FromResult(new HelloReply { Message = $"Hello {request.Name}" });
    }
}

ServerCallContext cung cấp context (ngữ cảnh) cho một server-side call (lời gọi phía server).

Service implementation được đăng ký với ứng dụng. Nếu service được host bởi ASP.NET Core gRPC, nó nên được thêm vào routing pipeline (đường ống định tuyến) với method MapGrpcService.

csharp

app.MapGrpcService<GreeterService>();

Xem gRPC services với ASP.NET Core để biết thêm thông tin.

Triển khai gRPC methods

Một gRPC service có thể có các loại method khác nhau. Cách messages được gửi và nhận bởi một service phụ thuộc vào loại method được định nghĩa. Các loại method gRPC là:

Unary (Đơn chiều)
Server streaming (Luồng từ server)
Client streaming (Luồng từ client)
Bi-directional streaming (Luồng hai chiều)

Streaming calls được chỉ định bằng từ khóa stream trong tệp .proto. stream có thể được đặt trên request message, response message, hoặc cả hai của một call.

protobuf

syntax = "proto3";

service ExampleService {
  // Unary
  rpc UnaryCall (ExampleRequest) returns (ExampleResponse);

  // Server streaming
  rpc StreamingFromServer (ExampleRequest) returns (stream ExampleResponse);

  // Client streaming
  rpc StreamingFromClient (stream ExampleRequest) returns (ExampleResponse);

  // Bi-directional streaming
  rpc StreamingBothWays (stream ExampleRequest) returns (stream ExampleResponse);
}

Mỗi loại call có một method signature (chữ ký phương thức) khác nhau. Ghi đè các method được tạo từ abstract base service type trong một concrete implementation đảm bảo các đối số và kiểu trả về đúng được sử dụng.

Unary method (Phương thức đơn chiều)

Unary method có request message là tham số và trả về response. Một unary call hoàn thành khi response được trả về.

csharp

public override Task<ExampleResponse> UnaryCall(ExampleRequest request,
    ServerCallContext context)
{
    var response = new ExampleResponse();
    return Task.FromResult(response);
}

Unary calls tương tự nhất với actions trên web API controllers. Một điểm khác biệt quan trọng là gRPC methods không thể bind (ràng buộc) các phần của request vào các đối số method khác nhau. gRPC methods luôn có một message argument cho dữ liệu request đến. Nhiều giá trị vẫn có thể được gửi đến gRPC service bằng cách thêm fields (trường) vào request message:

protobuf

message ExampleRequest {
    int32 pageIndex = 1;
    int32 pageSize = 2;
    bool isDescending = 3;
}

Server streaming method (Phương thức luồng từ server)

Server streaming method có request message là tham số. Vì nhiều messages có thể được stream (truyền) lại cho caller (người gọi), responseStream.WriteAsync được sử dụng để gửi response messages. Một server streaming call hoàn thành khi method trả về.

csharp

public override async Task StreamingFromServer(ExampleRequest request,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    for (var i = 0; i < 5; i++)
    {
        await responseStream.WriteAsync(new ExampleResponse());
        await Task.Delay(TimeSpan.FromSeconds(1));
    }
}

Client không có cách gửi thêm messages hoặc dữ liệu sau khi server streaming method đã bắt đầu. Một số streaming methods được thiết kế để chạy mãi mãi. Đối với continuous streaming methods (phương thức luồng liên tục), client có thể hủy call khi không còn cần thiết. Khi hủy xảy ra, client gửi tín hiệu đến server và ServerCallContext.CancellationToken được kích hoạt. CancellationToken nên được sử dụng trên server với async methods để:

Bất kỳ công việc bất đồng bộ nào được hủy cùng với streaming call.
Method thoát nhanh chóng.

csharp

public override async Task StreamingFromServer(ExampleRequest request,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    while (!context.CancellationToken.IsCancellationRequested)
    {
        await responseStream.WriteAsync(new ExampleResponse());
        await Task.Delay(TimeSpan.FromSeconds(1), context.CancellationToken);
    }
}

Client streaming method (Phương thức luồng từ client)

Client streaming method bắt đầu mà không có method nhận một message. Tham số requestStream được sử dụng để đọc messages từ client. Một client streaming call hoàn thành khi một response message được trả về:

csharp

public override async Task<ExampleResponse> StreamingFromClient(
    IAsyncStreamReader<ExampleRequest> requestStream, ServerCallContext context)
{
    await foreach (var message in requestStream.ReadAllAsync())
    {
        // ...
    }
    return new ExampleResponse();
}

Bi-directional streaming method (Phương thức luồng hai chiều)

Bi-directional streaming method bắt đầu mà không có method nhận một message. Tham số requestStream được sử dụng để đọc messages từ client. Method có thể chọn gửi messages với responseStream.WriteAsync. Một bi-directional streaming call hoàn thành khi method trả về:

csharp

public override async Task StreamingBothWays(IAsyncStreamReader<ExampleRequest> requestStream,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    await foreach (var message in requestStream.ReadAllAsync())
    {
        await responseStream.WriteAsync(new ExampleResponse());
    }
}

Mã trước:

Gửi một response cho mỗi request.
Là cách sử dụng cơ bản của bi-directional streaming.

Có thể hỗ trợ các kịch bản phức tạp hơn, chẳng hạn như đọc requests và gửi responses đồng thời:

csharp

public override async Task StreamingBothWays(IAsyncStreamReader<ExampleRequest> requestStream,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    // Read requests in a background task.
    var readTask = Task.Run(async () =>
    {
        await foreach (var message in requestStream.ReadAllAsync())
        {
            // Process request.
        }
    });

    // Send responses until the client signals that it is complete.
    while (!readTask.IsCompleted)
    {
        await responseStream.WriteAsync(new ExampleResponse());
        await Task.Delay(TimeSpan.FromSeconds(1), context.CancellationToken);
    }
}

Trong bi-directional streaming method, client và service có thể gửi messages cho nhau bất cứ lúc nào. Implementation tốt nhất của một bi-directional method thay đổi tùy thuộc vào yêu cầu.

Truy cập gRPC request headers

Request message không phải là cách duy nhất để client gửi dữ liệu đến gRPC service. Header values (giá trị tiêu đề) có sẵn trong service bằng cách sử dụng ServerCallContext.RequestHeaders.

csharp

public override Task<ExampleResponse> UnaryCall(ExampleRequest request,
    ServerCallContext context)
{
    var userAgent = context.RequestHeaders.GetValue("user-agent");
    // ...

    return Task.FromResult(new ExampleResponse());
}

Đa luồng với gRPC streaming methods

Có những lưu ý quan trọng khi triển khai gRPC streaming methods sử dụng nhiều threads (luồng).

Thread safety của Reader và Writer

IAsyncStreamReader<TMessage> và IServerStreamWriter<TMessage> chỉ có thể được sử dụng bởi một thread tại một thời điểm. Đối với streaming gRPC method, nhiều threads không thể đọc messages mới với requestStream.MoveNext() đồng thời. Và nhiều threads không thể viết messages mới với responseStream.WriteAsync(message) đồng thời.

Một cách an toàn để cho phép nhiều threads tương tác với một gRPC method là sử dụng producer-consumer pattern (mẫu nhà sản xuất-người tiêu dùng) với System.Threading.Channels.

csharp

public override async Task DownloadResults(DataRequest request,
        IServerStreamWriter<DataResult> responseStream, ServerCallContext context)
{
    var channel = Channel.CreateBounded<DataResult>(new BoundedChannelOptions(capacity: 5));

    var consumerTask = Task.Run(async () =>
    {
        // Consume messages from channel and write to response stream.
        await foreach (var message in channel.Reader.ReadAllAsync())
        {
            await responseStream.WriteAsync(message);
        }
    });

    var dataChunks = request.Value.Chunk(size: 10);

    // Write messages to channel from multiple threads.
    await Task.WhenAll(dataChunks.Select(
        async c =>
        {
            var message = new DataResult { BytesProcessed = c.Length };
            await channel.Writer.WriteAsync(message);
        }));

    // Complete writing and wait for consumer to complete.
    channel.Writer.Complete();
    await consumerTask;
}

gRPC server streaming method trước:

Tạo một bounded channel (kênh có giới hạn) để sản xuất và tiêu thụ DataResult messages.
Bắt đầu một task để đọc messages từ channel và viết chúng vào response stream.
Viết messages vào channel từ nhiều threads.

Lưu ý: Bidirectional streaming methods nhận IAsyncStreamReader<TMessage> và IServerStreamWriter<TMessage> làm đối số. Các kiểu này an toàn để sử dụng trên các threads riêng biệt.

Tương tác với gRPC method sau khi call kết thúc

Một gRPC call kết thúc trên server khi gRPC method thoát. Các đối số sau được truyền vào gRPC methods không an toàn để sử dụng sau khi call đã kết thúc:

ServerCallContext
IAsyncStreamReader<TMessage>
IServerStreamWriter<TMessage>

Nếu một gRPC method khởi động các background tasks (tác vụ nền) sử dụng các kiểu này, nó phải hoàn thành các tasks trước khi gRPC method thoát. Tiếp tục sử dụng context, stream reader, hoặc stream writer sau khi gRPC method tồn tại sẽ gây ra lỗi và hành vi không thể đoán trước.

Trong ví dụ sau, server streaming method có thể viết vào response stream sau khi call đã kết thúc:

csharp

public override async Task StreamingFromServer(ExampleRequest request,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    _ = Task.Run(async () =>
    {
        for (var i = 0; i < 5; i++)
        {
            await responseStream.WriteAsync(new ExampleResponse());
            await Task.Delay(TimeSpan.FromSeconds(1));
        }
    });

    await PerformLongRunningWorkAsync();
}

Đối với ví dụ trước, giải pháp là await (chờ) write task trước khi thoát method:

csharp

public override async Task StreamingFromServer(ExampleRequest request,
    IServerStreamWriter<ExampleResponse> responseStream, ServerCallContext context)
{
    var writeTask = Task.Run(async () =>
    {
        for (var i = 0; i < 5; i++)
        {
            await responseStream.WriteAsync(new ExampleResponse());
            await Task.Delay(TimeSpan.FromSeconds(1));
        }
    });

    await PerformLongRunningWorkAsync();

    await writeTask;
}