gRPC 기초 — Protocol Buffers부터 Spring Boot 연동까지

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@GrpcService
public class HelloGrpcService extends HelloServiceGrpc.HelloServiceImplBase {

    @Override
    public void sayHello(HelloRequest request,
                         StreamObserver<HelloResponse> responseObserver) {
        String message = String.format("안녕하세요, %s님! (나이: %d)",
                request.getName(), request.getAge());

        HelloResponse response = HelloResponse.newBuilder()
                .setMessage(message)
                .setTimestamp(System.currentTimeMillis())
                .build();

        responseObserver.onNext(response);     // 응답 전송
        responseObserver.onCompleted();         // 스트림 종료
    }

    @Override
    public void sayHelloStream(HelloRequest request,
                               StreamObserver<HelloResponse> responseObserver) {
        // Server Streaming: 3번에 걸쳐 응답 전송
        for (int i = 1; i <= 3; i++) {
            HelloResponse response = HelloResponse.newBuilder()
                    .setMessage(String.format("[%d/3] 안녕하세요, %s님!",
                            i, request.getName()))
                    .setTimestamp(System.currentTimeMillis())
                    .build();

            responseObserver.onNext(response);

            try {
                Thread.sleep(1000); // 1초 간격으로 전송
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                responseObserver.onError(e);
                return;
            }
        }
        responseObserver.onCompleted();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@Service
public class HelloGrpcClient {

    private final HelloServiceGrpc.HelloServiceBlockingStub blockingStub;

    public HelloGrpcClient(
            @GrpcClient("hello-service")
            HelloServiceGrpc.HelloServiceBlockingStub blockingStub) {
        this.blockingStub = blockingStub;
    }

    // Unary 호출
    public String sayHello(String name, int age) {
        HelloRequest request = HelloRequest.newBuilder()
                .setName(name)
                .setAge(age)
                .build();

        HelloResponse response = blockingStub.sayHello(request);
        return response.getMessage();
    }

    // Server Streaming 호출
    public List<String> sayHelloStream(String name) {
        HelloRequest request = HelloRequest.newBuilder()
                .setName(name)
                .build();

        List<String> messages = new ArrayList<>();
        Iterator<HelloResponse> responses =
                blockingStub.sayHelloStream(request);

        while (responses.hasNext()) {
            messages.add(responses.next().getMessage());
        }
        return messages;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@RestController
@RequestMapping("/api/hello")
@RequiredArgsConstructor
public class HelloController {

    private final HelloGrpcClient helloGrpcClient;

    @GetMapping("/{name}")
    public ResponseEntity<Map<String, String>> hello(
            @PathVariable String name,
            @RequestParam(defaultValue = "0") int age) {

        String message = helloGrpcClient.sayHello(name, age);
        return ResponseEntity.ok(Map.of("message", message));
    }

    @GetMapping("/{name}/stream")
    public ResponseEntity<List<String>> helloStream(
            @PathVariable String name) {

        List<String> messages = helloGrpcClient.sayHelloStream(name);
        return ResponseEntity.ok(messages);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@GrpcService
public class ChatGrpcService extends ChatServiceGrpc.ChatServiceImplBase {

    private final Set<StreamObserver<ChatMessage>> clients =
            ConcurrentHashMap.newKeySet();

    @Override
    public StreamObserver<ChatMessage> chat(
            StreamObserver<ChatMessage> responseObserver) {

        clients.add(responseObserver);

        return new StreamObserver<>() {
            @Override
            public void onNext(ChatMessage message) {
                // 모든 클라이언트에 브로드캐스트
                for (StreamObserver<ChatMessage> client : clients) {
                    if (client != responseObserver) {
                        client.onNext(message);
                    }
                }
            }

            @Override
            public void onError(Throwable t) {
                clients.remove(responseObserver);
            }

            @Override
            public void onCompleted() {
                clients.remove(responseObserver);
                responseObserver.onCompleted();
            }
        };
    }
}

1
2
3
4
5
6
7
8
9
10
11
@Override
public void sayHello(HelloRequest request,
                     StreamObserver<HelloResponse> responseObserver) {
    if (request.getName().isBlank()) {
        responseObserver.onError(Status.INVALID_ARGUMENT
                .withDescription("이름은 필수입니다.")
                .asRuntimeException());
        return;
    }
    // 정상 처리...
}

1
2
3
HelloResponse response = blockingStub
        .withDeadlineAfter(3, TimeUnit.SECONDS)  // 3초 타임아웃
        .sayHello(request);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@GrpcGlobalServerInterceptor
public class LoggingInterceptor implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {

        String method = call.getMethodDescriptor().getFullMethodName();
        long start = System.currentTimeMillis();

        ServerCall.Listener<ReqT> listener = next.startCall(call, headers);

        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<>(
                listener) {
            @Override
            public void onComplete() {
                long elapsed = System.currentTimeMillis() - start;
                log.info("gRPC {} — {}ms", method, elapsed);
                super.onComplete();
            }
        };
    }
}