Comienza a usar gRPC-Go: transmisión

1. Introducción

En este codelab, usarás gRPC-Go para crear un cliente y un servidor que formen la base de una aplicación de asignación de rutas escrita en Go.

Al final del instructivo, tendrás un cliente que se conecta a un servidor remoto con gRPC para obtener información sobre las funciones en la ruta de un cliente, crear un resumen de la ruta de un cliente y compartir información de la ruta, como actualizaciones de tráfico, con el servidor y otros clientes.

El servicio se define en un archivo de Protocol Buffers, que se usará para generar código estándar para el cliente y el servidor, de modo que puedan comunicarse entre sí, lo que te ahorrará tiempo y esfuerzo en la implementación de esa funcionalidad.

Este código generado se encarga no solo de las complejidades de la comunicación entre el servidor y el cliente, sino también de la serialización y deserialización de datos.

Qué aprenderás

  • Cómo usar los búferes de protocolo para definir una API de servicio
  • Cómo compilar un cliente y un servidor basados en gRPC a partir de una definición de Protocol Buffers con la generación de código automatizada
  • Conocimiento de la comunicación de transmisión cliente-servidor con gRPC

Este codelab está dirigido a desarrolladores de Go que no conocen gRPC o que desean repasar sus conceptos, o a cualquier otra persona interesada en crear sistemas distribuidos. No se requiere experiencia previa con gRPC.

2. Antes de comenzar

Requisitos previos

Asegúrate de haber instalado lo siguiente:

  • La versión 1.24.5 o posterior de la cadena de herramientas de Go Para obtener instrucciones de instalación, consulta la sección Comenzar de Go.
  • El compilador de búferes de protocolo, protoc, versión 3.27.1 o posterior Para obtener instrucciones de instalación, consulta la guía de instalación del compilador.
  • Son los complementos del compilador de búferes de protocolo para Go y gRPC. Para instalar estos complementos, ejecuta los siguientes comandos:
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

Actualiza tu variable PATH para que el compilador de búferes de protocolo pueda encontrar los complementos:

export PATH="$PATH:$(go env GOPATH)/bin"

Obtén el código

Para que no tengas que empezar desde cero, este codelab proporciona un andamio del código fuente de la aplicación para que lo completes. En los siguientes pasos, se muestra cómo finalizar la aplicación, incluido el uso de los complementos del compilador de búfer de protocolo para generar el código gRPC estándar.

Primero, crea el directorio de trabajo del codelab y cd en él:

mkdir streaming-grpc-go-getting-started && cd streaming-grpc-go-getting-started

Descarga y extrae el codelab:

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-1/codelabs/grpc-go-streaming/start_here

También puedes descargar el archivo .zip que contiene solo el directorio del codelab y descomprimirlo de forma manual.

El código fuente completo está disponible en GitHub si no quieres escribir una implementación.

3. Define mensajes y servicios

El primer paso es definir el servicio de gRPC de la aplicación, sus métodos de RPC y sus tipos de mensajes de solicitud y respuesta con búferes de protocolo. Tu servicio proporcionará lo siguiente:

  • Métodos de RPC llamados ListFeatures, RecordRoute y RouteChat que el servidor implementa y el cliente llama.
  • Los tipos de mensajes Point, Feature, Rectangle, RouteNote y RouteSummary, que son estructuras de datos que se intercambian entre el cliente y el servidor cuando se llaman a los métodos anteriores.

Estos métodos de RPC y sus tipos de mensajes se definirán en el archivo routeguide/route_guide.proto del código fuente proporcionado.

Los búferes de protocolo se conocen comúnmente como protobufs. Para obtener más información sobre la terminología de gRPC, consulta los conceptos básicos, la arquitectura y el ciclo de vida de gRPC.

Define los tipos de mensajes

En el archivo routeguide/route_guide.proto del código fuente, primero define el tipo de mensaje Point. Un objeto Point representa un par de coordenadas de latitud y longitud en un mapa. En este codelab, usa números enteros para las coordenadas:

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

Los números 1 y 2 son números de ID únicos para cada uno de los campos de la estructura message.

A continuación, define el tipo de mensaje Feature. Un Feature usa un campo string para el nombre o la dirección postal de algo en una ubicación especificada por un Point:

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

A continuación, un mensaje Rectangle que representa un rectángulo de latitud y longitud, representado como dos puntos opuestos diagonalmente "lo" y "hi".

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

También es un mensaje de RouteNote que representa un mensaje enviado en un punto determinado.

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

También requeriríamos un mensaje de RouteSummary. Este mensaje se recibe en respuesta a una RPC de RecordRoute, que se explica en la siguiente sección. Contiene la cantidad de puntos individuales recibidos, la cantidad de atributos detectados y la distancia total recorrida como la suma acumulativa de la distancia entre cada punto.

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

Cómo definir métodos de servicio

Para definir un servicio, especifica un servicio con nombre en tu archivo .proto. El archivo route_guide.proto tiene una estructura service llamada RouteGuide que define uno o más métodos proporcionados por el servicio de la aplicación.

Define métodos RPC dentro de la definición de tu servicio y especifica sus tipos de solicitud y respuesta. En esta sección del codelab, definiremos lo siguiente:

ListFeatures

Obtiene los Feature disponibles en el Rectangle determinado. Los resultados se transmiten en lugar de devolverse de una vez (p.ej., en un mensaje de respuesta con un campo repetido), ya que el rectángulo puede abarcar un área grande y contener una gran cantidad de entidades.

Un tipo adecuado para esta RPC es una RPC de transmisión del servidor: el cliente envía una solicitud al servidor y obtiene una transmisión para leer una secuencia de mensajes. El cliente lee la transmisión que se muestra hasta que no haya más mensajes. Como puedes ver en nuestro ejemplo, para especificar un método de transmisión del servidor, debes colocar la palabra clave stream antes del tipo de respuesta.

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

Acepta un flujo de Points en una ruta que se recorre y devuelve un RouteSummary cuando se completa el recorrido.

En este caso, parece apropiada una RPC de transmisión del cliente: el cliente escribe una secuencia de mensajes y los envía al servidor, nuevamente a través de una transmisión proporcionada. Una vez que el cliente termina de escribir los mensajes, espera a que el servidor los lea todos y muestre la respuesta. Para especificar un método de transmisión por Internet del cliente, coloca la palabra clave stream antes del tipo de solicitud.

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

Acepta un flujo de objetos RouteNote enviados mientras se recorre una ruta y recibe otros objetos RouteNote (p.ej., de otros usuarios).

Este es exactamente el tipo de caso de uso para la transmisión bidireccional. En una RPC de transmisión bidireccional, ambos extremos envían una secuencia de mensajes a través de una transmisión de lectura y escritura. Las dos transmisiones operan de forma independiente, por lo que los clientes y los servidores pueden leer y escribir en el orden que deseen.

Por ejemplo, el servidor podría esperar a recibir todos los mensajes del cliente antes de escribir sus respuestas, o bien podría leer un mensaje y, luego, escribir uno, o alguna otra combinación de lecturas y escrituras.

Se conserva el orden de los mensajes en cada transmisión. Para especificar este tipo de método, coloca la palabra clave stream antes de la solicitud y la respuesta.

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. Genera código de cliente y servidor

A continuación, genera el código gRPC estándar para el cliente y el servidor desde el archivo .proto con el compilador de búfer de protocolo. En el directorio routeguide, ejecuta lo siguiente:

protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       route_guide.proto

Este comando genera los siguientes archivos:

  • route_guide.pb.go, que contiene funciones para crear los tipos de mensajes de la aplicación y acceder a sus datos, y la definición de los tipos que representan los mensajes.
  • route_guide_grpc.pb.go, que contiene funciones que el cliente usa para llamar al método gRPC remoto del servicio y funciones que el servidor usa para proporcionar ese servicio remoto.

A continuación, implementaremos los métodos del servidor para que, cuando el cliente envíe una solicitud, el servidor pueda responder con una respuesta.

5. Implementa el servicio

Primero, veamos cómo crear un servidor RouteGuide. Para que nuestro servicio de RouteGuide funcione correctamente, se deben realizar dos pasos:

  • Implementar la interfaz de servicio generada a partir de nuestra definición de servicio: realizar el "trabajo" real de nuestro servicio
  • Ejecutar un servidor de gRPC para escuchar las solicitudes de los clientes y enviarlas a la implementación de servicio correcta

Implementemos RouteGuide en server/server.go.

Implementa RouteGuide

Debemos implementar la interfaz RouteGuideService generada. Así se vería la implementación.

type routeGuideServer struct {
        ...
}
...
func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
        ...
}
...

func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
        ...
}
...

func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
        ...
}

Analicemos en detalle cada implementación de RPC.

RPC de transmisión del servidor

Comienza con uno de nuestros RPC de transmisión. ListFeatures es una RPC de transmisión del servidor, por lo que debemos enviar varios Features a nuestro cliente.

func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
  for _, feature := range s.savedFeatures {
    if inRange(feature.Location, rect) {
      if err := stream.Send(feature); err != nil {
        return err
      }
    }
  }
  return nil
}

Como puedes ver, en lugar de obtener objetos de solicitud y respuesta simples en los parámetros de nuestro método, esta vez obtenemos un objeto de solicitud (el Rectangle en el que nuestro cliente quiere encontrar Features) y un objeto RouteGuide_ListFeaturesServer especial para escribir nuestras respuestas. En el método, completamos tantos objetos Feature como necesitamos devolver y los escribimos en RouteGuide_ListFeaturesServer con su método Send(). Por último, al igual que en nuestro RPC simple, devolvemos un error nil para indicarle a gRPC que terminamos de escribir respuestas. Si se produce algún error en esta llamada, devolvemos un error no nulo. La capa de gRPC lo traducirá a un estado de RPC adecuado para enviarlo por cable.

RPC de transmisión del cliente

Ahora veamos algo un poco más complicado: el método de transmisión del cliente RecordRoute, en el que obtenemos un flujo de Points del cliente y devolvemos un solo RouteSummary con información sobre su viaje. Como puedes ver, esta vez el método no tiene ningún parámetro de solicitud. En su lugar, obtiene una transmisión RouteGuide_RecordRouteServer, que el servidor puede usar para leer y escribir mensajes. Puede recibir mensajes del cliente con su método Recv() y devolver su única respuesta con su método SendAndClose().

func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
  var pointCount, featureCount, distance int32
  var lastPoint *pb.Point
  startTime := time.Now()
  for {
    point, err := stream.Recv()
    if err == io.EOF {
      endTime := time.Now()
      return stream.SendAndClose(&pb.RouteSummary{
        PointCount:   pointCount,
        FeatureCount: featureCount,
        Distance:     distance,
        ElapsedTime:  int32(endTime.Sub(startTime).Seconds()),
      })
    }
    if err != nil {
      return err
    }
    pointCount++
    for _, feature := range s.savedFeatures {
      if proto.Equal(feature.Location, point) {
        featureCount++
      }
    }
    if lastPoint != nil {
      distance += calcDistance(lastPoint, point)
    }
    lastPoint = point
  }
}

En el cuerpo del método, usamos el método Recv() de RouteGuide_RecordRouteServer para leer repetidamente las solicitudes de nuestro cliente en un objeto de solicitud (en este caso, un Point) hasta que no haya más mensajes: el servidor debe verificar el error que devuelve Recv() después de cada llamada. Si es nil, el flujo sigue siendo válido y se puede seguir leyendo. Si es io.EOF, el flujo de mensajes finalizó y el servidor puede devolver su RouteSummary. Si tiene cualquier otro valor, devolvemos el error "tal cual" para que la capa de gRPC lo traduzca a un estado de RPC.

RPC de transmisión bidireccional

Por último, veamos nuestro RPC de transmisión bidireccional RouteChat().

func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
  for {
    in, err := stream.Recv()
    if err == io.EOF {
      return nil
    }
    if err != nil {
      return err
    }
    key := serialize(in.Location)

    s.mu.Lock()
    s.routeNotes[key] = append(s.routeNotes[key], in)
    // Note: this copy prevents blocking other clients while serving this one.
    // We don't need to do a deep copy, because elements in the slice are
    // insert-only and never modified.
    rn := make([]*pb.RouteNote, len(s.routeNotes[key]))
    copy(rn, s.routeNotes[key])
    s.mu.Unlock()

    for _, note := range rn {
      if err := stream.Send(note); err != nil {
        return err
      }
    }
  }
}

Esta vez, obtenemos una transmisión de RouteGuide_RouteChatServer que, como en nuestro ejemplo de transmisión del cliente, se puede usar para leer y escribir mensajes. Sin embargo, esta vez devolvemos valores a través de la transmisión de nuestro método mientras el cliente sigue escribiendo mensajes en su transmisión de mensajes. La sintaxis para leer y escribir aquí es muy similar a nuestro método de transmisión del cliente, excepto que el servidor usa el método send() de la transmisión en lugar de SendAndClose() porque escribe varias respuestas. Aunque cada lado siempre recibirá los mensajes del otro en el orden en que se escribieron, tanto el cliente como el servidor pueden leer y escribir en cualquier orden. Las transmisiones operan de forma completamente independiente.

Inicia el servidor

Una vez que implementamos todos nuestros métodos, también debemos iniciar un servidor de gRPC para que los clientes puedan usar nuestro servicio. En el siguiente fragmento, se muestra cómo lo hacemos para nuestro servicio RouteGuide:

lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", *port))
if err != nil {
  log.Fatalf("failed to listen: %v", err)
}

grpcServer := grpc.NewServer()

s := &routeGuideServer{routeNotes: make(map[string][]*pb.RouteNote)}
s.loadFeatures()
pb.RegisterRouteGuideServer(grpcServer, s)
grpcServer.Serve(lis)

Esto es lo que sucede en main(), paso a paso:

  1. Especifica el puerto TCP que se usará para escuchar las solicitudes de clientes remotos con lis, err := net.Listen(...). De forma predeterminada, la aplicación usa el puerto TCP 50051, como se especifica en la variable port o pasando el parámetro --port en la línea de comandos cuando se ejecuta el servidor. Si no se puede abrir el puerto TCP, la aplicación finaliza con un error fatal.
  2. Crea una instancia del servidor de gRPC con grpc.NewServer(...) y asígnale el nombre grpcServer.
  3. Crea un puntero a routeGuideServer, una estructura que representa el servicio de la API de la aplicación, y nómbralo s..
  4. Usa s.loadFeatures() para propagar el array s.savedFeatures.
  5. Registra nuestra implementación del servicio con el servidor de gRPC.
  6. Llama a Serve() en el servidor con los detalles de nuestro puerto para realizar una espera de bloqueo de las solicitudes del cliente. Esto continúa hasta que se detiene el proceso o se llama a Stop().

La función loadFeatures() obtiene sus asignaciones de coordenadas a ubicación de server/testdata.go.

6. Crea el cliente

Ahora, edita client/client.go, que es donde implementarás el código del cliente.

Para llamar a los métodos del servicio remoto, primero debemos crear un canal de gRPC para comunicarnos con el servidor. Para ello, pasamos la cadena URI de destino del servidor (que, en este caso, es simplemente la dirección y el número de puerto) a grpc.NewClient() en la función main() del cliente de la siguiente manera:

// Set up a connection to the gRPC server.
conn, err := grpc.NewClient("dns:///"+*serverAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
  log.Fatalf("fail to dial: %v", err)
}
defer conn.Close()

La dirección del servidor, definida por la variable serverAddr, es localhost:50051 de forma predeterminada y se puede anular con el parámetro --addr en la línea de comandos cuando se ejecuta el cliente.

Si el cliente necesita conectarse a un servicio que requiere credenciales de autenticación, como credenciales de TLS o JWT, puede pasar un objeto DialOptions como parámetro a grpc.NewClient que contenga las credenciales requeridas. El servicio RouteGuide no requiere credenciales.

Una vez que se configura el canal de gRPC, necesitamos un stub del cliente para realizar RPC a través de llamadas a funciones de Go. Obtenemos ese stub con el método NewRouteGuideClient que proporciona el archivo route_guide_grpc.pb.go generado a partir del archivo .proto de la aplicación.

import (pb "github.com/grpc-ecosystem/codelabs/getting_started_streaming/routeguide")

client := pb.NewRouteGuideClient(conn)

Llama a los métodos de servicio

Ahora veamos cómo llamamos a nuestros métodos de servicio. En gRPC-Go, las RPC operan en un modo de bloqueo o síncrono, lo que significa que la llamada a la RPC espera a que responda el servidor y mostrará una respuesta o un error.

RPC de transmisión del servidor

Aquí es donde llamamos al método de transmisión del servidor ListFeatures, que devuelve un flujo de objetos geográficos Feature.

rect := &pb.Rectangle{ ... }  // initialize a pb.Rectangle
log.Printf("Looking for features within %v", rect)
stream, err := client.ListFeatures(context.Background(), rect)
if err != nil {
  log.Fatalf("client.ListFeatures failed: %v", err)
}
for {
  // For server-to-client streaming RPCs, you call stream.Recv() until it
  // returns io.EOF.
  feature, err := stream.Recv()
  if err == io.EOF {
    break
  }
  if err != nil {
    log.Fatalf("client.ListFeatures failed: %v", err)
  }
  log.Printf("Feature: name: %q, point:(%v, %v)", feature.GetName(),
    feature.GetLocation().GetLatitude(), feature.GetLocation().GetLongitude())
}

Al igual que en la RPC simple, le pasamos al método un contexto y una solicitud. Sin embargo, en lugar de obtener un objeto de respuesta, obtenemos una instancia de RouteGuide_ListFeaturesClient. El cliente puede usar la transmisión RouteGuide_ListFeaturesClient para leer las respuestas del servidor. Usamos el método Recv() de RouteGuide_ListFeaturesClient para leer repetidamente las respuestas del servidor a un objeto de búfer de protocolo de respuesta (en este caso, un Feature) hasta que no haya más mensajes: el cliente debe verificar el error err que se devuelve de Recv() después de cada llamada. Si es nil, el flujo sigue siendo válido y se puede seguir leyendo. Si es io.EOF, el flujo de mensajes finalizó. De lo contrario, debe haber un error de RPC, que se pasa a través de err.

RPC de transmisión del cliente

El método de transmisión del cliente RecordRoute es similar al método del servidor, excepto que solo pasamos un contexto al método y obtenemos una transmisión RouteGuide_RecordRouteClient, que podemos usar para escribir y leer mensajes.

// Create a random number of random points
r := rand.New(rand.NewSource(time.Now().UnixNano()))
pointCount := int(r.Int31n(100)) + 2 // Traverse at least two points
var points []*pb.Point
for i := 0; i < pointCount; i++ {
  points = append(points, randomPoint(r))
}
log.Printf("Traversing %d points.", len(points))
c2sStream, err := client.RecordRoute(context.TODO())
if err != nil {
  log.Fatalf("client.RecordRoute failed: %v", err)
}
// Stream each point to the server.
for _, point := range points {
  if err := c2sStream.Send(point); err != nil {
    log.Fatalf("client.RecordRoute: stream.Send(%v) failed: %v", point, err)
  }
}
// Close the stream and receive the RouteSummary from the server.
reply, err := c2sStream.CloseAndRecv()
if err != nil {
  log.Fatalf("client.RecordRoute failed: %v", err)
}
log.Printf("Route summary: %v", reply)

El objeto RouteGuide_RecordRouteClient tiene un método Send() que podemos usar para enviar solicitudes al servidor. Una vez que terminamos de escribir las solicitudes de nuestro cliente en la transmisión con Send(), debemos llamar a CloseAndRecv() en la transmisión para que gRPC sepa que terminamos de escribir y que esperamos recibir una respuesta. Obtenemos el estado de la RPC del error que devuelve CloseAndRecv(). Si el estado es nil, el primer valor de devolución de CloseAndRecv() será una respuesta válida del servidor.

RPC de transmisión bidireccional

Por último, veamos nuestro RPC de transmisión bidireccional RouteChat(). Al igual que en el caso de RecordRoute, solo pasamos al método un objeto de contexto y obtenemos un flujo que podemos usar para escribir y leer mensajes. Sin embargo, esta vez devolvemos valores a través de la transmisión de nuestro método mientras el servidor sigue escribiendo mensajes en su transmisión de mensajes.

biDiStream, err := client.RouteChat(context.Background())
if err != nil {
  log.Fatalf("client.RouteChat failed: %v", err)
}
// this channel is used to wait for the receive goroutine to finish.
recvDoneCh := make(chan struct{})
// receive goroutine.
go func() {
  for {
    in, err := biDiStream.Recv()
    if err == io.EOF {
      // read done.
      close(recvDoneCh)
      return
    }
    if err != nil {
      log.Fatalf("client.RouteChat failed: %v", err)
    }
    log.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude)
  }
}()
// send messages simultaneously.
for _, note := range notes {
  if err := biDiStream.Send(note); err != nil {
    log.Fatalf("client.RouteChat: stream.Send(%v) failed: %v", note, err)
  }
}
biDiStream.CloseSend()
// wait for the receive goroutine to finish.
<-recvDoneCh

La sintaxis para leer y escribir aquí es muy similar a nuestro método de transmisión del cliente, excepto que usamos el método CloseSend() de la transmisión una vez que finalizamos nuestra llamada. Aunque cada lado siempre recibirá los mensajes del otro en el orden en que se escribieron, tanto el cliente como el servidor pueden leer y escribir en cualquier orden. Las transmisiones operan de forma completamente independiente.

7. Probar

Ejecuta los siguientes comandos en el directorio de trabajo de la aplicación para confirmar que el servidor y el cliente funcionen correctamente entre sí:

  1. Ejecuta el servidor en una terminal:
cd server
go run .
  1. Ejecuta el cliente desde otra terminal:
cd client
go run .

Verás un resultado como este, con marcas de tiempo omitidas para mayor claridad:

Looking for features within lo:<latitude:400000000 longitude:-750000000 > hi:<latitude:420000000 longitude:-730000000 >
name:"Patriots Path, Mendham, NJ 07945, USA" location:<latitude:407838351 longitude:-746143763 >
...
name:"3 Hasta Way, Newton, NJ 07860, USA" location:<latitude:410248224 longitude:-747127767 >
Traversing 56 points.
Route summary: point_count:56 distance:497013163
Got message First message at point(0, 1)
Got message Second message at point(0, 2)
Got message Third message at point(0, 3)
Got message First message at point(0, 1)
Got message Fourth message at point(0, 1)
Got message Second message at point(0, 2)
Got message Fifth message at point(0, 2)
Got message Third message at point(0, 3)
Got message Sixth message at point(0, 3)

8. ¿Qué sigue?