1. 程式人生 > >Dapr 運用之整合 Asp.Net Core Grpc 呼叫篇

Dapr 運用之整合 Asp.Net Core Grpc 呼叫篇

前置條件: 《Dapr 運用》


改造 ProductService 以提供 gRPC 服務

  1. 從 NuGet 或程式包管理控制檯安裝 gRPC 服務必須的包

    • Grpc.AspNetCore
  2. 配置 Http/2
    • gRPC 服務需要 Http/2 協議

      public static IHostBuilder CreateHostBuilder(string[] args)
      {
          return Host.CreateDefaultBuilder(args)
              .ConfigureWebHostDefaults(webBuilder =>
              {
                  webBuilder.ConfigureKestrel(options =>
                  {
                      options.Listen(IPAddress.Loopback, 50![](https://img2018.cnblogs.com/blog/757544/201912/757544-20191218205830077-211023565.png)
      01, listenOptions =>
                      {
                          listenOptions.Protocols = HttpProtocols.Http2;
                      });
                  });
                  webBuilder.UseStartup<Startup>();
              });
      }
  3. 新建了 product.proto 以定義 GRPC 服務,它需要完成的內容是返回所有產品集合,當然目前產品內容只有一個 ID

    • 定義產品 proto

      syntax = "proto3";
      
      package productlist.v1;
      
      option csharp_namespace = "ProductList.V1";
      
      service ProductRPCService{
          rpc GetAllProducts(ProductListRequest) returns(ProductList);
      }
      
      message ProductListRequest{
      
      }
      
      message ProductList {
          repeated Product results = 1;
      }
      
      message Product {
          string ID=1;
      }
      說明
      • 定義產品列表 gRPC 服務,得益於宇宙第一 IDE Visual Studio ,只要新增 Grpc.Tools 包就可以自動生成 gRPC 所需的程式碼,這裡不再需要手動去新增 Grpc.Tools ,官方提供的 Grpc.AspNetCore 中已經集成了
      • 定義了一個服務 ProductRPCService
      • 定義了一個函式 ProductRPCService
      • 定義了一個請求構造 ProductListRequest ,內容為空
      • 定義了一個請求返回構造 ProductList ,使用 repeated 表明返回資料是集合
      • 定義了一個數據集合中的一個物件 Product
    • 新增 ProductListService 檔案,內容如下

          public class ProductListService : ProductRPCService.ProductRPCServiceBase
          {
              private readonly ProductContext _productContext;
      
              public ProductListService(ProductContext productContext)
              {
                  _productContext = productContext;
              }
      
              public override async Task<ProductList.V1.ProductList> GetAllProducts(ProductListRequest request, ServerCallContext context)
              {
                  IList<Product> results = await _productContext.Products.ToListAsync();
                  var productList = new ProductList.V1.ProductList();
                  foreach (Product item in results)
                  {
                      productList.Results.Add(new ProductList.V1.Product
                      {
                          ID = item.ProductID.ToString()
                      });
                  }
      
                  return productList;
              }
          }
  4. 在 Startup.cs 修改程式碼如下

    public void ConfigureServices(IServiceCollection services)
    {
        //啟用 gRPC 服務
        services.AddGrpc();
        services.AddTransient<ProductListService>();
        ...
    }

    這裡的 services.AddTransient

    public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
    {
        if (env.IsDevelopment())
        {
            app.UseDeveloperExceptionPage();
        }
    
        app.UseRouting();
    
        app.UseEndpoints(endpoints =>
        {
            ...
    
            //新增 gRPC 到路由管道中
            endpoints.MapGrpcService<DaprClientService>();
        });
    }

    這裡新增的程式碼的含義分別是啟用 gRPC 服務和新增 gRPC 路由。得益於 ASP.NET Core 中介軟體的優秀設計,ASP.NET Core 可同時支援 Http 服務。

  5. 新增 daprclient.proto 檔案以生成 Dapr Grpc 服務,daprclient.proto 內容如下

    syntax = "proto3";
    
    package daprclient;
    
    import "google/protobuf/any.proto";
    import "google/protobuf/empty.proto";
    import "google/protobuf/duration.proto";
    
    option java_outer_classname = "DaprClientProtos";
    option java_package = "io.dapr";
    
    // User Code definitions
    service DaprClient {
    rpc OnInvoke (InvokeEnvelope) returns (google.protobuf.Any) {}
    rpc GetTopicSubscriptions(google.protobuf.Empty) returns (GetTopicSubscriptionsEnvelope) {}
    rpc GetBindingsSubscriptions(google.protobuf.Empty) returns (GetBindingsSubscriptionsEnvelope) {}
    rpc OnBindingEvent(BindingEventEnvelope) returns (BindingResponseEnvelope) {}
    rpc OnTopicEvent(CloudEventEnvelope) returns (google.protobuf.Empty) {}
    }
    
    message CloudEventEnvelope {
    string id = 1;
    string source = 2;
    string type = 3;
    string specVersion = 4;
    string dataContentType = 5;
    string topic = 6;
    google.protobuf.Any data = 7;
    }
    
    message BindingEventEnvelope {
        string name = 1;
        google.protobuf.Any data = 2;
        map<string,string> metadata = 3;
    }
    
    message BindingResponseEnvelope {
    google.protobuf.Any data = 1;
    repeated string to = 2;
    repeated State state = 3;
    string concurrency = 4;
    }
    
    message InvokeEnvelope {
        string method = 1;
        google.protobuf.Any data = 2;
        map<string,string> metadata = 3;
    }
    
    message GetTopicSubscriptionsEnvelope {
    repeated string topics = 1;
    }
    
    message GetBindingsSubscriptionsEnvelope {
    repeated string bindings = 1;
    }
    
    message State {
    string key = 1;
    google.protobuf.Any value = 2;
    string etag = 3;
    map<string,string> metadata = 4;
    StateOptions options = 5;
    }
    
    message StateOptions {
    string concurrency = 1;
    string consistency = 2;
    RetryPolicy retryPolicy = 3;
    }
    
    message RetryPolicy {
    int32 threshold = 1;
    string pattern = 2;
    google.protobuf.Duration interval = 3;
    }
    說明
    • 此檔案為官方提供,Dapr 0.3 版本之前提供的已經生成好的程式碼,現在看原始碼可以看出已經改為提供 proto 檔案了,這裡我認為提供 proto 檔案比較合理
    • 此檔案定義了5個函式,此文主要講的就是 OnInvoke() 函式
    • OnInvoke() 請求構造為 InvokeEnvelope
      • method 提供呼叫方法名稱
      • data 請求資料
      • metadata 額外資料,此處使用鍵值對形式體現
  6. 建立 DaprClientService.cs 檔案,此檔案用於終結點路由,內容為

    public class DaprClientService : DaprClient.DaprClientBase
    {
        private readonly ProductListService _productListService;
    
        /// <summary>
        /// Initializes a new instance of the <see cref="ProductService" /> class.
        /// </summary>
        /// <param name="productListService"></param>
        public DaprClientService(ProductListService productListService)
        {
            _productListService = productListService;
        }
    
        public override async Task<Any> OnInvoke(InvokeEnvelope request, ServerCallContext context)
        {
            switch (request.Method)
            {
                case "GetAllProducts":
                    ProductListRequest productListRequest = ProductListRequest.Parser.ParseFrom(request.Data.Value);
                    ProductList.V1.ProductList productsList = await _productListService.GetAllProducts(productListRequest, context);
                    return Any.Pack(productsList);
            }
    
            return null;
        }
    }
    說明
    • 使用構造器注入已定義好的 ProductListService
    • InvokeEnvelope 中的 Method 用於路由資料
    • 使用 ProductListRequest.Parser.ParseFrom 轉換請求構造
    • 使用 Any.Pack() 打包需要返回的資料
  7. 執行 productService

    dapr run --app-id productService --app-port 5001 --protocol grpc dotnet run

小結
至此,ProductService 服務完成。此時 ProductService.Api.csproj Protobuf 內容為

<ItemGroup>
  <Protobuf Include="Protos\daprclient.proto" GrpcServices="Server" />
 <Protobuf Include="Protos\productList.proto" GrpcServices="Server" />
 </ItemGroup>

改造 StorageService 服務以完成 Dapr GRPC 服務呼叫

  1. 新增 productList.proto 檔案,內容同 ProductService 中的 productList.proto

  2. 新增 dapr.proto 檔案,此檔案也為官方提供,內容為

    syntax = "proto3";
    
    package dapr;
    
    import "google/protobuf/any.proto";
    import "google/protobuf/empty.proto";
    import "google/protobuf/duration.proto";
    
    option java_outer_classname = "DaprProtos";
    option java_package = "io.dapr";
    
    option csharp_namespace = "Dapr.Client.Grpc";
    
    
    // Dapr definitions
    service Dapr {
    rpc PublishEvent(PublishEventEnvelope) returns (google.protobuf.Empty) {}
    rpc InvokeService(InvokeServiceEnvelope) returns (InvokeServiceResponseEnvelope) {}
    rpc InvokeBinding(InvokeBindingEnvelope) returns (google.protobuf.Empty) {}
    rpc GetState(GetStateEnvelope) returns (GetStateResponseEnvelope) {}
    rpc SaveState(SaveStateEnvelope) returns (google.protobuf.Empty) {}
    rpc DeleteState(DeleteStateEnvelope) returns (google.protobuf.Empty) {}
    }
    
    message InvokeServiceResponseEnvelope {
    google.protobuf.Any data = 1;
    map<string,string> metadata = 2;
    }
    
    message DeleteStateEnvelope {
    string key = 1;
    string etag = 2;
    StateOptions options = 3;
    }
    
    message SaveStateEnvelope {
    repeated StateRequest requests = 1;
    }
    
    message GetStateEnvelope {
        string key = 1;
        string consistency = 2;
    }
    
    message GetStateResponseEnvelope {
    google.protobuf.Any data = 1;
    string etag = 2;
    }
    
    message InvokeBindingEnvelope {
    string name = 1;
    google.protobuf.Any data = 2;
    map<string,string> metadata = 3;
    }
    
    message InvokeServiceEnvelope {
    string id = 1;
    string method = 2;
    google.protobuf.Any data = 3;
    map<string,string> metadata = 4;
    }
    
    message PublishEventEnvelope {
        string topic = 1;
        google.protobuf.Any data = 2;
    }
    
    message State {
    string key = 1;
    google.protobuf.Any value = 2;
    string etag = 3;
    map<string,string> metadata = 4;
    StateOptions options = 5;
    }
    
    message StateOptions {
    string concurrency = 1;
    string consistency = 2;
    RetryPolicy retryPolicy = 3;
    }
    
    message RetryPolicy {
    int32 threshold = 1;
    string pattern = 2;
    google.protobuf.Duration interval = 3;
    }
    
    message StateRequest {
    string key = 1;
    google.protobuf.Any value = 2;
    string etag = 3;
    map<string,string> metadata = 4;
    StateRequestOptions options = 5;
    }
    
    message StateRequestOptions {
    string concurrency = 1;
    string consistency = 2;
    StateRetryPolicy retryPolicy = 3;
    }
    
    message StateRetryPolicy {
    int32 threshold = 1;
    string pattern = 2;
    google.protobuf.Duration interval = 3;
    }
    說明
    • 此檔案提供6個 GRPC 服務,此文介紹的函式為 InvokeService()
      • 請求構造為 InvokeServiceEnvelope
        • id 請求的服務的 --app-id ,比如 productService
        • method 請求的方法
        • data 請求函式的簽名
        • metadata 元資料鍵值對
  3. 修改 StorageController 中的 InitialStorage() 函式為

    /// <summary>
    /// 初始化倉庫.
    /// </summary>
    /// <returns>是否成功.</returns>
    [HttpGet("InitialStorage")]
    public async Task<bool> InitialStorage()
    {
        string defaultPort = Environment.GetEnvironmentVariable("DAPR_GRPC_PORT") ?? "5001";
    
        // Set correct switch to make insecure gRPC service calls. This switch must be set before creating the GrpcChannel.
        AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
    
        // Create Client
        string daprUri = $"http://127.0.0.1:{defaultPort}";
        GrpcChannel channel = GrpcChannel.ForAddress(daprUri);
        var client = new Dapr.Client.Grpc.Dapr.DaprClient(channel);
    
        InvokeServiceResponseEnvelope result = await client.InvokeServiceAsync(new InvokeServiceEnvelope
        {
            Method = "GetAllProducts",
            Id = "productService",
            Data = Any.Pack(new ProductListRequest())
        });
        ProductList.V1.ProductList productResult = ProductList.V1.ProductList.Parser.ParseFrom(result.Data.Value);
    
        var random = new Random();
    
        foreach (Product item in productResult.Results)
        {
            _storageContext.Storage.Add(new Storage
            {
                ProductID = Guid.Parse(item.ID),
                Amount = random.Next(1, 1000)
            });
        }
    
        await _storageContext.SaveChangesAsync();
        return true;
    }
  4. 啟動 StorageService

    dapr run --app-id storageService --app-port 5003 dotnet run
  5. 使用 Postman 請求 StorageService 的 InitialStorage

  6. 使用 MySql Workbench 檢視結果

小結
至此,以 Dapr 框架使用 GRPC 客戶端在 StorageService 中完成了對 ProductService 服務的呼叫。

原始碼地