gRPC 与微服务通信

// 一个最小的 Proto 契约
syntax = "proto3";

package order;

service OrderService {
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc CreateOrder (CreateOrderRequest) returns (Order);
}

message GetOrderRequest {
  int32 id = 1;
}

message Order {
  int32 id = 1;
  string customer = 2;
  double total = 3;
  repeated OrderLine lines = 4;
}

message OrderLine {
  int32 product_id = 1;
  int32 quantity = 2;
}

syntax = "proto3";

package shop.orders.v1;   // 生成命名空间 Shop.Orders.V1

import "google/protobuf/timestamp.proto";

message Order {
  int32 id = 1;
  string customer = 2;
  double total = 3;
  google.protobuf.Timestamp created_at = 4;
  repeated OrderLine lines = 5;

  reserved 6;            // 已删除字段，编号不可复用
  reserved "legacy_flag";
}

service OrderApi {
  rpc ListOrders (ListOrdersRequest) returns (ListOrdersResponse);
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc WatchOrders (WatchOrdersRequest) returns (stream Order); // 服务端流
}

<ItemGroup>
  <Protobuf Include="Protos\orders.proto" GrpcServices="Both" />
</ItemGroup>

// Program.cs —— 注册 gRPC 服务
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddGrpc();
var app = builder.Build();

app.MapGrpcService<OrderApiService>();

app.Run();

// 服务实现：继承生成的基类
public class OrderApiService(IOrderRepository repo) : OrderApi.OrderApiBase
{
    public override async Task<Order> GetOrder(GetOrderRequest request, ServerCallContext context)
    {
        var order = await repo.GetByIdAsync(request.Id, context.CancellationToken);
        return order is null
            ? throw new RpcException(new Status(StatusCode.NotFound, "order not found"))
            : MapToProto(order);
    }
}

// 客户端调用
using var channel = GrpcChannel.ForAddress("https://orders.internal:5001");
var client = new OrderApi.OrderApiClient(channel);

var reply = await client.GetOrderAsync(new GetOrderRequest { Id = 42 });
Console.WriteLine(reply.Customer);

// 服务端拦截器：日志 + 耗时
public class LoggingInterceptor : Interceptor
{
    private readonly ILogger<LoggingInterceptor> _logger;

    public LoggingInterceptor(ILogger<LoggingInterceptor> logger) => _logger = logger;

    public override async Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
        TRequest request, ServerCallContext context,
        UnaryServerMethod<TRequest, TResponse> continuation)
    {
        var sw = Stopwatch.StartNew();
        try
        {
            return await continuation(request, context);
        }
        catch (RpcException) { throw; }
        catch (Exception ex)
        {
            throw new RpcException(new Status(StatusCode.Internal, ex.Message));
        }
        finally
        {
            sw.Stop();
            _logger.LogInformation("gRPC {Method} took {ElapsedMs}ms",
                context.Method, sw.ElapsedMilliseconds);
        }
    }
}

// 注册拦截器
builder.Services.AddGrpc(options =>
{
    options.Interceptors.Add<LoggingInterceptor>();
});

// 服务端流：边查边推
public override async Task WatchOrders(
    WatchOrdersRequest request,
    IServerStreamWriter<Order> responseStream,
    ServerCallContext context)
{
    await foreach (var order in _orderFeed.WatchAsync(context.CancellationToken))
    {
        await responseStream.WriteAsync(order); // 推送一条
    }
}

// 双向流：客户端不断发、服务器不断回
public override async Task Chat(
    IAsyncStreamReader<ChatMessage> requestStream,
    IServerStreamWriter<ChatReply> responseStream,
    ServerCallContext context)
{
    await foreach (var msg in requestStream.ReadAllAsync(context.CancellationToken))
    {
        var reply = new ChatReply { Text = $"echo: {msg.Text}" };
        await responseStream.WriteAsync(reply);
    }
}

// 引入 Grpc.AspNetCore.HealthChecks
builder.Services.AddGrpcHealthChecks()
    .AddCheck("orders", () => HealthCheckResult.Healthy());

app.MapGrpcHealthChecksService(); // 暴露标准 /grpc.health.v1.Health

// YARP 反向代理配置示例（appsettings.json 片段）
// 把 /api/orders/** 转发到 gRPC 服务，做 HTTP/1.1 到 HTTP/2 的翻译

builder.Services.AddGrpc(options =>
{
    options.MaxReceiveMessageSize = 4 * 1024 * 1024; // 4MB
    options.MaxSendMessageSize = 4 * 1024 * 1024;
    options.EnableDetailedErrors = builder.Environment.IsDevelopment();
    options.Interceptors.Add<AuthInterceptor>();
});

// 认证：把客户端证书或 token 转为 ClaimsPrincipal
public class AuthInterceptor : Interceptor
{
    public override async Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
        TRequest request, ServerCallContext context,
        UnaryServerMethod<TRequest, TResponse> continuation)
    {
        var token = context.RequestHeaders.GetValue("authorization")?
                            .Replace("Bearer ", "");
        if (!_validator.IsValid(token))
            throw new RpcException(new Status(StatusCode.Unauthenticated, "bad token"));
        return await continuation(request, context);
    }
}