GRPC-services en -methoden maken

Opmerking

Dit is niet de nieuwste versie van dit artikel. Zie de .NET 10-versie van dit artikel voor de huidige release.

Waarschuwing

Deze versie van ASP.NET Core wordt niet meer ondersteund. Zie het .NET- en .NET Core-ondersteuningsbeleid voor meer informatie. Zie de .NET 10-versie van dit artikel voor de huidige release.

Door James Newton-King

In dit document wordt uitgelegd hoe u gRPC-services en -methoden maakt in C#. Onderwerpen zijn onder andere:

  • Services en methoden definiëren in .proto bestanden.
  • Gegenereerde code met gRPC C#-hulpprogramma's.
  • GRPC-services en -methoden implementeren.

Nieuwe gRPC-services maken

gRPC-services met C# hebben gRPC's contract-first benadering voor API-ontwikkeling geïntroduceerd. Services en berichten worden gedefinieerd in .proto bestanden. C#-hulpprogramma's genereren vervolgens code uit .proto bestanden. Voor assets aan de serverzijde wordt een abstract basistype gegenereerd voor elke service, samen met klassen voor berichten.

Het volgende .proto bestand:

  • Hiermee definieert u een Greeter service.
  • De Greeter service definieert een SayHello aanroep.
  • SayHello verzendt een HelloRequest bericht en ontvangt een HelloReply bericht
syntax = "proto3";

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

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

Met C#-hulpprogramma's wordt het C# GreeterBase -basistype gegenereerd:

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; }
}

De gegenereerde GreeterBase opdracht doet standaard niets. De virtuele SayHello methode retourneert een UNIMPLEMENTED fout bij clients die deze aanroepen. Opdat de service nuttig is, moet een app een concrete implementatie van GreeterBase creëren.

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

De ServerCallContext functie geeft de context voor een aanroep aan de serverzijde.

De service-implementatie wordt geregistreerd bij de app. Als de service wordt gehost door ASP.NET Core gRPC, moet deze worden toegevoegd aan de routeringspijplijn met de MapGrpcService methode.

app.MapGrpcService<GreeterService>();

Zie gRPC-services met ASP.NET Core voor meer informatie.

GRPC-methoden implementeren

Een gRPC-service kan verschillende soorten methoden hebben. Hoe berichten door een service worden verzonden en ontvangen, is afhankelijk van het type methode dat is gedefinieerd. De gRPC-methodetypen zijn:

  • Unary
  • Server streaming
  • Clientstreaming
  • Bidirectioneel streamen

Streaming-aanroepen worden opgegeven met het stream trefwoord in het .proto bestand. stream kan worden geplaatst op het aanvraagbericht, het antwoordbericht of beide van een oproep.

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);
}

Elk aanroeptype heeft een andere methodehandtekening. Het overschrijven van gegenereerde methoden van het abstracte basisservicetype in een concrete implementatie zorgt ervoor dat de juiste argumenten en het retourtype worden gebruikt.

Unaire methode

Een unaire methode heeft het aanvraagbericht als parameter en retourneert het antwoord. Een unaire aanroep is voltooid wanneer het antwoord wordt geretourneerd.

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

Unary-aanroepen zijn het meest vergelijkbaar met acties op web-API-controllers. Een belangrijk verschil dat gRPC-methoden hebben van acties, is dat gRPC-methoden geen delen van een aanvraag kunnen binden aan verschillende methodeargumenten. gRPC-methoden hebben altijd één berichtargument voor de binnenkomende aanvraaggegevens. Meerdere waarden kunnen nog steeds worden verzonden naar een gRPC-service door velden toe te voegen aan het aanvraagbericht:

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

Serverstreamingsmethode

Een serverstreamingmethode heeft het aanvraagbericht als parameter. Omdat meerdere berichten naar de beller kunnen worden gestreamd, responseStream.WriteAsync wordt gebruikt om antwoordberichten te verzenden. Een aanroep voor serverstreaming is voltooid wanneer de methode wordt geretourneerd.

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));
    }
}

De client kan geen extra berichten of gegevens verzenden zodra de serverstreamingmethode is gestart. Sommige streamingmethoden zijn ontworpen om voor altijd te worden uitgevoerd. Voor continue streamingmethoden kan een client de aanroep annuleren wanneer deze niet meer nodig is. Wanneer annulering plaatsvindt, stuurt de client een signaal naar de server en wordt het ServerCallContext.CancellationToken gegenereerd. Het CancellationToken token moet worden gebruikt op de server met asynchrone methoden, zodat:

  • Asynchroon werk wordt samen met de streaming-aanroep geannuleerd.
  • De methode eindigt snel.
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);
    }
}

Clientstreamingsmethode

Een clientstreamingmethode wordt gestart zonder de methode die een bericht ontvangt. De requestStream parameter wordt gebruikt om berichten van de client te lezen. Een clientstreaming-aanroep is voltooid wanneer een antwoordbericht wordt geretourneerd:

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

Bidirectionele streamingmethode

Een bidirectionele streamingmethode wordt gestart zonder dat de methode een bericht ontvangt. De requestStream parameter wordt gebruikt om berichten van de client te lezen. De methode kan ervoor kiezen om berichten te verzenden met responseStream.WriteAsync. Een bidirectionele streaming-aanroep is voltooid wanneer de methode retourneert:

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());
    }
}

De voorgaande code:

  • Hiermee wordt een antwoord verzonden voor elke aanvraag.
  • Is een basisgebruik van bidirectionele streaming.

Het is mogelijk om complexere scenario's te ondersteunen, zoals leesaanvragen en het tegelijkertijd verzenden van antwoorden:

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);
    }
}

In een bidirectionele streamingmethode kunnen de client en service op elk gewenst moment berichten naar elkaar verzenden. De beste implementatie van een bidirectionele methode is afhankelijk van de vereisten.

Toegang tot gRPC-aanvraagheaders

Een aanvraagbericht is niet de enige manier voor een client om gegevens naar een gRPC-service te verzenden. Headerwaarden zijn beschikbaar in een service met behulp van ServerCallContext.RequestHeaders.

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

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

Multithreading met gRPC-streamingmethoden

Er zijn belangrijke overwegingen voor het implementeren van gRPC-streamingmethoden die meerdere threads gebruiken.

Veiligheid van lezer en schrijverthread

IAsyncStreamReader<TMessage> en IServerStreamWriter<TMessage> kan elk worden gebruikt door slechts één thread tegelijk. Voor een streaming gRPC-methode kunnen meerdere threads geen nieuwe berichten requestStream.MoveNext() tegelijk lezen. Meerdere threads kunnen niet tegelijkertijd nieuwe berichten met responseStream.WriteAsync(message) schrijven.

Een veilige manier om meerdere threads in staat te stellen om te communiceren met een gRPC-methode is door het patroon producer-consumer te gebruiken met System.Threading.Channels.

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;
}

De voorgaande streamingmethode voor gRPC-servers:

  • Hiermee maakt u een gebonden kanaal voor het produceren en gebruiken van DataResult berichten.
  • Hiermee start een taak om berichten uit het kanaal te lezen en deze naar de antwoordstroom te schrijven.
  • Hiermee worden berichten vanuit meerdere threads naar het kanaal geschreven.

Opmerking

Bidirectionele streamingmethoden nemen IAsyncStreamReader<TMessage> en IServerStreamWriter<TMessage> als argumenten. Het is veilig om deze typen op afzonderlijke threads te gebruiken.

Interactie met een gRPC-methode nadat een aanroep is beëindigd

Een gRPC-aanroep eindigt op de server zodra de gRPC-methode wordt afgesloten. De volgende argumenten die worden doorgegeven aan gRPC-methoden zijn niet veilig om te gebruiken nadat de aanroep is beëindigd:

  • ServerCallContext
  • IAsyncStreamReader<TMessage>
  • IServerStreamWriter<TMessage>

Als een gRPC-methode achtergrondtaken start die deze typen gebruiken, moet deze de taken voltooien voordat de gRPC-methode wordt afgesloten. Als u de context, streamlezer of stream writer blijft gebruiken nadat de gRPC-methode bestaat, worden fouten en onvoorspelbaar gedrag veroorzaakt.

In het volgende voorbeeld kan de streamingmethode van de server naar de antwoordstroom schrijven nadat de aanroep is voltooid:

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();
}

Bij het vorige voorbeeld moet u wachten op de voltooiing van de schrijftaak voordat u de methode afsluit.

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;
}

Aanvullende bronnen

  • Last updated on