gRPC

gRPC

• google에서 개발한 오픈소스 RPC(Remote Procedure Call) 프레임워크
• HTTP/2를 사용한다.
  • 한 커넥션으로 동시에 여러 개의 메시지를 주고 받을 수 있다.
• client application은 다른 컴퓨터의 server application에서 메서드를 직접 호출할 수 있다. ➡️ 분산 응용 프로그램 및 서비스를 쉽게 만든다.
• 원격으로 호출될 파라미터와 리턴 타입을 가지는 메서드를 작성한다.
• 서버 측에서 서버는 이 인터페이스를 구현하고 클라이언트 호출을 다루기 위해 gRPC 서버를 실행한다.
• 클라이언트 측에서 클라이언트에는 서버와 동일한 메서드를 제공하는 stub를 가진다.
  • 서버와 클라이언트는 서로 다른 주소 공간을 사용하므로, 함수 호출에 사용된 매개 변수를 꼭 변환해줘야 한다.
• gRPC 클라이언트와 서버는 다양한 환경에서 실행되고 서로 통신할 수 있으며 gRPC가 지원하는 언어로 작성할 수 있다.
  • ex) 자바로 작성된 gRPC 서버와 Go, Python, Ruby로 작성된 클라이언트
• Protocol Buffers
  • 기본적으로 gRPC는 Protocol Buffers를 사용한다.
  • google에서 개발한 구조화된 데이터를 직렬화(바이트 단위로 변환하는 작업)하는 기법
  • proto 파일을 작성하자.

 

 

의존성

runtimeOnly 'io.grpc:grpc-netty-shaded:1.52.1'
implementation 'io.grpc:grpc-protobuf:1.52.1'
implementation 'io.grpc:grpc-stub:1.52.1'
compileOnly 'org.apache.tomcat:annotations-api:6.0.53' // necessary for Java 9+

gRPC -Java사용을 위해 추가한다.

 

 

proto

• Message and Field
  • 주고 받는 데이터를 message로 정의

syntax = "proto3";

message SearchRequest {
  string query = 1;
  int32 page_number = 2;
  int32 result_per_page = 3;
}

• Field Number
  • 메시지에 정의된 필드들은 각각 고유한 번호를 가진다.
  • Encoding 이후 binary data에서 필드를 식별하는데 사용
  • 1 ~ 536,870,911
    • ⚠️ 19000 ~ 19999은 reserved 된 값이라서 사용 불가
• Proto File Field Rule
  • singular
    • 해당 필드를 가지지 않거나 하나만 가진다.
    • proto3를 사용할 때, 필드에 다른 field rule이 없다면 singular가 default field rule이다.
  • optional
    • 해당 필드를 가지지 않거나 하나만 가진다.
  • repeated
    • 순서가 유지되는 임의 반복 가능한 필드
    • proto3에서 scalar numeric 타입의 repeated 필드는 압축 인코딩을 기본으로 사용한다.
  • map
    • key, value 쌍

 

 

 

service

• gRPC 서비스와 protocol buffers를 사용해서 메소드 request, response 타입을 정의

service RouteGuide {
   ...
}

.proto 파일 안에 service를 생성

 

// Obtains the feature at a given position.
rpc GetFeature(Point) returns (Feature) {}

위의 서비스 안에 rpc 메서드를 정의한다.

클라이언트가 stub을 이용해서 어떤 서버로 요청을 보내고 응답이 돌아오기를 기다린다.

 

// Obtains the Features available within the given Rectangle.  Results are
// streamed rather than returned at once (e.g. in a response message with a
// repeated field), as the rectangle may cover a large area and contain a
// huge number of features.
rpc ListFeatures(Rectangle) returns (stream Feature) {}

 서버측 streaming RPC는 클라이언트가 서버로 요청을 보내고 stream을 받는다.

클라이언트가 서버에 요청을 보내고 일련의 메시지를 다시 읽기 위한 스트림을 가져오는 서버 측 스트리밍 RPC다.

클라이언트는 메시지가 더이상 없을 때까지 반환되는 stream을 읽는다.

위의 예제에서 response 타입 앞에 stream 키워드를 붙였다.

 

// Accepts a stream of Points on a route being traversed, returning a
// RouteSummary when traversal is completed.
rpc RecordRoute(stream Point) returns (RouteSummary) {}

클라이언트 측 streaming RPC는 클라이언트가 일련의 메시지를 쓰고 주어진 stream을 다시 이용해서 서버에 보낸다.

일단 클라이언트가 메시지를 쓰는 것을 마치면, 서버가 이것을 다 읽을 때까지 기다리고 응답을 보낸다.

request 타입 앞에 stream을 붙인다.

 

// Accepts a stream of RouteNotes sent while a route is being traversed,
// while receiving other RouteNotes (e.g. from other users).
rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

 양방향 streaming RPC

두 stream은 독립적으로 작동한다.

클라이언트와 서버는 읽기, 쓰기를 원하는 순서대로 할 수 있다.

예를 들어, 서버는 응답이 작성되기 전에 모든 클라이언트 메시지를 기다리거나, 메시지를 읽은 후 메시지를 작성하거나, 읽기와 쓰기의 조합을 사용할 수 있다.

각 stream의 메시지의 순서는 유지된다.

 

 

// Points are represented as latitude-longitude pairs in the E7 representation
// (degrees multiplied by 10**7 and rounded to the nearest integer).
// Latitudes should be in the range +/- 90 degrees and longitude should be in
// the range +/- 180 degrees (inclusive).
message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

 .proto 파일은 서비스 메소드에서 사용되는 모든 request, response 타입에 대한 protocol buffer message type 정의를 포함한다.

위의 예제는 Point 메시지 타입을 나타낸다.

 

 

client

서비스 메서드를 호출하기 위해 stub을 생성해야 한다.

두 가지 방법이 있다.

1. a blocking/synchronous stub : RPC 호출은 서버의 응답을 기다리고, 응답을 반환하거나 예외를 발생시키게 된다.
2. a non-blocking/asynchronous stub : 응답이 비동기식으로 반환되는 서버에 non-blocking 요청을 수행한다. 특정 유형의 streaming 호출은 비동기 stub만을 사용하여 수행할 수 있다.

// https://grpc.io/docs/languages/java/basics/
public RouteGuideClient(String host, int port) {
  this(ManagedChannelBuilder.forAddress(host, port).usePlaintext());
}

/** Construct client for accessing RouteGuide server using the existing channel. */
public RouteGuideClient(ManagedChannelBuilder<?> channelBuilder) {
  channel = channelBuilder.build();
  blockingStub = RouteGuideGrpc.newBlockingStub(channel);
  asyncStub = RouteGuideGrpc.newStub(channel);
}

 ManagedChannelBuilder를 사용해서 channel을 생성했다.

채널을 사용해서 stub을 생성할 수 있다.

 

// https://grpc.io/docs/languages/java/basics/
Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();
Feature feature;
try {
  feature = blockingStub.getFeature(request);
} catch (StatusRuntimeException e) {
  logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
  return;
}

Simple RPC

에러가 발생하면, Status로 인코딩된다.

 

 

// https://grpc.io/docs/languages/java/basics/
Rectangle request =
    Rectangle.newBuilder()
        .setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build())
        .setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build()).build();
Iterator<Feature> features;
try {
  features = blockingStub.listFeatures(request);
} catch (StatusRuntimeException e) {
  logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
  return;
}

 Server-side streaming RPC

ListFeatures는 Features의 stream을 반환

메소드는 클라이언트가 반환된 모든 Features를 읽을 수 있는 Iterator를 반환

 

 

public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
  info("*** RecordRoute");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {
    @Override
    public void onNext(RouteSummary summary) {
      info("Finished trip with {0} points. Passed {1} features. "
          + "Travelled {2} meters. It took {3} seconds.", summary.getPointCount(),
          summary.getFeatureCount(), summary.getDistance(), summary.getElapsedTime());
    }

    @Override
    public void onError(Throwable t) {
      Status status = Status.fromThrowable(t);
      logger.log(Level.WARNING, "RecordRoute Failed: {0}", status);
      finishLatch.countDown();
    }

    @Override
    public void onCompleted() {
      info("Finished RecordRoute");
      finishLatch.countDown();
    }
  };

  StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
  try {
    // Send numPoints points randomly selected from the features list.
    Random rand = new Random();
    for (int i = 0; i < numPoints; ++i) {
      int index = rand.nextInt(features.size());
      Point point = features.get(index).getLocation();
      info("Visiting point {0}, {1}", RouteGuideUtil.getLatitude(point),
          RouteGuideUtil.getLongitude(point));
      requestObserver.onNext(point);
      // Sleep for a bit before sending the next one.
      Thread.sleep(rand.nextInt(1000) + 500);
      if (finishLatch.getCount() == 0) {
        // RPC completed or errored before we finished sending.
        // Sending further requests won't error, but they will just be thrown away.
        return;
      }
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

Client-side streaming RPC

client-side streaming 메소드 RecordRoute는 Points의 스트림을 서버로 보내고 단일 RouteSummary을 다시 얻는다.

이 메소드에서 비동기 stub을 사용할 필요가 있다.

이 메소드를 호출하기 위해 서버가 RouteSummary 응답으로 호출할 수 있는 특수 인터페이스를 구현하는 StreamObserver를 생성해야 한다.

• 서버가 RouteSummary를 메시지 stream에 쓸때 반환된 정보를 출력하기 위해 onNext() 메서드를 재정의한다.
• 서버가 쓰기를 완료했는지 체크하는 CountDownLatch를 줄이기 위해 onCompleted() 메서드를(서버가 호출을 완료했을 때 호출됨)  재정의한다.

StreamObserver를 비동기 stub의 recordRoute() 메소드에 전달하고 자체 StreamObserver 요청 관찰자를 다시 가져와 서버에 보낼 포인트를 작성한다.

points 작성이 끝나면 요청 관찰자의 onCompleted() 메서드를 사용하여 gRPC에 클라이언트 측에서 작성이 완료되었음을 알려준다.

작업이 완료되면 CountDownLatch를 확인하여 서버가 측면에서 완료되었는지 확인한다.

 

 

public void routeChat() throws Exception {
  info("*** RoutChat");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteNote> requestObserver =
      asyncStub.routeChat(new StreamObserver<RouteNote>() {
        @Override
        public void onNext(RouteNote note) {
          info("Got message \"{0}\" at {1}, {2}", note.getMessage(), note.getLocation()
              .getLatitude(), note.getLocation().getLongitude());
        }

        @Override
        public void onError(Throwable t) {
          Status status = Status.fromThrowable(t);
          logger.log(Level.WARNING, "RouteChat Failed: {0}", status);
          finishLatch.countDown();
        }

        @Override
        public void onCompleted() {
          info("Finished RouteChat");
          finishLatch.countDown();
        }
      });

  try {
    RouteNote[] requests =
        {newNote("First message", 0, 0), newNote("Second message", 0, 1),
            newNote("Third message", 1, 0), newNote("Fourth message", 1, 1)};

    for (RouteNote request : requests) {
      info("Sending message \"{0}\" at {1}, {2}", request.getMessage(), request.getLocation()
          .getLatitude(), request.getLocation().getLongitude());
      requestObserver.onNext(request);
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

 Bidirectional streaming RPC

각 측은 항상 작성된 순서대로 상대방의 메시지를 받는다.

클라이언트와 서버는 모든 순서로 읽고 쓸 수 있다.

stream은 완전히 독립적으로 작동한다.

 

 

참고 👇

https://grpc.io/docs/what-is-grpc/introduction/

Introduction to gRPC

 

https://github.com/grpc/grpc-java/blob/master/README.md

GitHub - grpc/grpc-java: The Java gRPC implementation. HTTP/2 based RPC

 

https://protobuf.dev/programming-guides/proto3/

Language Guide (proto 3)

 

https://medium.com/naver-cloud-platform/nbp-%EA%B8%B0%EC%88%A0-%EA%B2%BD%ED%97%98-%EC%8B%9C%EB%8C%80%EC%9D%98-%ED%9D%90%EB%A6%84-grpc-%EA%B9%8A%EA%B2%8C-%ED%8C%8C%EA%B3%A0%EB%93%A4%EA%B8%B0-1-39e97cb3460

[NBP 기술&경험] 시대의 흐름, gRPC 깊게 파고들기 #1

 

https://yidongnan.github.io/grpc-spring-boot-starter/en/server/getting-started.html

Getting Started

 

https://erjuer.tistory.com/116

[MSA] gRPC 프레임 워크 적용기