gRPC 통신 패턴

RPC(Remote Procedure Call)

• 네트워크로 연결된 서버 상의 프로시저를 원격으로 호출할 수 있는 기능

• IDL(Interface Definication Language) 기반으로 다양한 언어 환경에서 쉽게 확장 가능
• C++, Java, Python, Node.Js, Go, C# 등..

• Stub
• RPC 호출을 위해서는 Stub이 꼭 필요하다.
• Stub은 프로시저 호출을 추상화하는 작은 코드 조각이다.
• 클라이언트/서버 모두에 존재하며, 호출 및 반환 과정을 처리하는 역할을 한다.
• 클라이언트→marshal , 서버→Unmarshal 로 데이터 직렬화

• RPC의 통신 과정
1. IDL를 사용하여 통신 규약을 정의한다
grpc의 경우 proto로 stub code 생성
2. 만들어진 stub 코드를 클라이언트/서버에 함께 빌드
3. 클라이언트는 stub 사용 → RPC 런타임을 통해 함수 호출
4. 서버는 프로시저 호출에 대한 처리 후 결과 값 반환
→ 클라이언트는 서버의 결과 값을 반환받고 함수를 local에 있는 것처럼 사용할 수 있다.

• RPC를 사용하는 이유? (REST와 비교해서)
1. 성능 - 일반적으로 REST보다 적은 오버헤드를 가진다.(직렬화)
2. 실시간 통신 - RPC는 양방향 통신을 지원하여 실시간 스트리밍이 필요한 경우 유용하다.
3. 다양한 언어 지원 - 다양한 언어로의 애플리케이션 통신을 용이하게 한다.
4. 엄격한 인터페이스 정의 - IDL(ex-proto)를 사용하여 정의하기 때문에, 클라이언트-서버간 일관성 유지에 도움이 된다.
 

GRPC

• Google에서 개발한 오픈소스 RPC 프레임워크이다.

• 기존 Protocol Buffer(매세지 직렬화) 기반에 HTTP/2를 결합하여 새 RPC 프레임워크를 만듬
• HTTP/2
• 한 커넥션으로 동시에 여러개의 메세지를 주고 받을 수 있으며 response는 순서에 상관없이 stream으로 주고받는다.
HTTP 1.0, 1.1, 2.0
• Protocol Buffer(proto, protobuf)
• Serialization : 데이터를 바이트 단위로 변환하는 것
• json text형식 데이터를 Serialization하면 용량 압축 가능
• 사용을 위해 protocol buffer 기본 정보를 명시하는 proto file이 필요하다.
• 예시 proto file (go)

syntax = "proto3";

package main;

// 이 옵션에 맞게 Go 패키지(모듈) 구조 생성
option go_package = "./grpc";

// 서비스 -> 서비스 안에 들어갈 function 인터페이스 정의
service ProductInfo {
    rpc addProduct(Product) returns (ProductID);
    rpc getProduct(ProductID) returns (Product);
}

// 메세지 -> 데이터 필드 명시
message Product {
    string id = 1;
    string name = 2;
    string description = 3;
    float price = 4;
}

message ProductID {
    string value = 1;
}

• Setup
• protocol compiler가 설치되어 있는지 확인

$ go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28
$ go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.2

• protoc 플러그인 PATH 업데이트
• $ export PATH="$PATH:$(go env GOPATH)/bin"
• proto file을 작성하고 protobuf 컴파일러 명령어를 날려주면, .pb.go 파일이 생성됨
• test.proto 파일을 상대경로로 현재 폴더에 변환
• → test.pb.go, test_grpc.pb.go 파일이 생성

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

• 해당 파일을 통해 grpc를 통한 서비스 로직 구현
‣

// 단순 RPC
func (s *routeGuideServer) GetFeature(ctx context.Context, point *pb.Point) (*pb.Feature, error) {
        ...
}

// 서버측 스트리밍 RPC
func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
        ...
}

// 클라이언트측 스트리밍 RPC
func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
        ...
}

// 양방향 스트리밍 RPC
func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
        ...
}

• GRPC 통신 패턴
1. 단순 RPC(단일 패턴)
• 클라이언트가 서버에 요청을 보내면 응답을 받는, 간단한 RPC

// 서버
func (s *routeGuideServer) GetFeature(ctx context.Context, point *pb.Point) (*pb.Feature, error) {
	for _, feature := range s.savedFeatures {
		if proto.Equal(feature.Location, point) {
			return feature, nil
		}
	}
	// No feature was found, return an unnamed feature
	return &pb.Feature{Location: point}, nil
}

// 클라이언트 
func printFeature(client pb.RouteGuideClient, point *pb.Point) {
	feature, err := client.GetFeature(ctx, point)
	if err != nil {
		log.Fatalf("client.GetFeature failed: %v", err)
	}
}

2. 서버 측 스트리밍 RPC
• 클라이언트가 서버에 요청을 보내고, 서버로부터 더 이상 메세지가 없을 때까지 스트림을 읽음
• 서버 Send() & 클라이언트 Recv()

// 서버
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
}

// 클라이언트
rect := &pb.Rectangle{ ... }  // initialize a pb.Rectangle
stream, err := client.ListFeatures(context.Background(), rect)
if err != nil {
  ...
}
for {
    feature, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Fatalf("%v.ListFeatures(_) = _, %v", client, err)
    }
    log.Println(feature)
}

3. 클라이언트 측 스트리밍 RPC
• 클라이언트는 메세지를 작성하고, 서버가 메세지를 모두 읽고 응답을 반환할 때까지 기다린다.
• 서버 Recv() & 클라이언트 Send()
• 스트리밍하는 클라이언트의 메세지를 수신하여, 응답을 반환함

// 서버
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
  }
}

// 클라이언트
// 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))
stream, err := client.RecordRoute(context.Background())
if err != nil {
  log.Fatalf("%v.RecordRoute(_) = _, %v", client, err)
}
for _, point := range points {
  if err := stream.Send(point); err != nil {
    log.Fatalf("%v.Send(%v) = %v", stream, point, err)
  }
}
reply, err := stream.CloseAndRecv()
if err != nil {
  log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
}
log.Printf("Route summary: %v", reply)

4. 양방향 스트리밍 RPC
• 양쪽에서 읽기-쓰기 스트림을 사용하여 일련의 메시지를 보냄
• 두 스트림은 독립적으로 작동한다.
• 서버 Send() and Recv() & 클라이언트 Recv() and Send()

// 서버
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)
                ... // look for notes to be sent to client
    for _, note := range s.routeNotes[key] {
      if err := stream.Send(note); err != nil {
        return err
      }
    }
  }
}

// 클라이언트
stream, err := client.RouteChat(context.Background())
waitc := make(chan struct{})
go func() {
  for {
    in, err := stream.Recv()
    if err == io.EOF {
      // read done.
      close(waitc)
      return
    }
    if err != nil {
      log.Fatalf("Failed to receive a note : %v", err)
    }
    log.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude)
  }
}()
for _, note := range notes {
  if err := stream.Send(note); err != nil {
    log.Fatalf("Failed to send a note: %v", err)
  }
}
stream.CloseSend()
<-waitc

번외 - JSON-RPC

• RPC를 JSON 방식으로 표현한 것

• 요청 메세지 작성
• jsonrpc - JSON RPC 버전 명시
• method - 호출하려는 원격 프로시저의 이름
• params - 원격 프로시저에 전달할 매개변수
• id - 요청을 식별하기 위한 고유 식별자

• 이더리움에서 JSON-RPC를 통해 이더리움 노드와 통신한다. → web3.js
• ex

// Request
curl -X POST --data '{"jsonrpc":"2.0","method":"eth_call","params":[{see above}],"id":1}'
// Result
{
  "id":1,
  "jsonrpc": "2.0",
  "result": "0x"
}