gRPC-Java의 Netty 활용

// 서비스와 메시지를 .proto 파일에 정의
syntax = "proto3";

service GreeterService {
  // Unary RPC — 요청 하나, 응답 하나
  rpc SayHello (HelloRequest) returns (HelloReply);

  // Server Streaming — 요청 하나, 응답 여러 개
  rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

// gRPC 클라이언트 채널 생성
ManagedChannel channel = ManagedChannelBuilder
    .forAddress("localhost", 50051)
    .usePlaintext()  // 개발 환경에서만 사용 (TLS 비활성화)
    .build();

// 스텁 생성 — 이 스텁이 실제 RPC 호출을 수행
GreeterServiceGrpc.GreeterServiceBlockingStub stub =
    GreeterServiceGrpc.newBlockingStub(channel);

// RPC 호출 — 메서드 호출처럼 사용
HelloReply reply = stub.sayHello(
    HelloRequest.newBuilder().setName("Netty").build()
);

ManagedChannel
├── NameResolver (DNS, 정적 주소 등)
├── LoadBalancer (pick-first, round-robin)
└── Subchannel[]
    ├── Subchannel → Netty Channel (서버 A:50051)
    └── Subchannel → Netty Channel (서버 B:50051)
        └── Bootstrap으로 생성된 TCP 연결
            └── ChannelPipeline
                ├── SslHandler (TLS)
                ├── Http2FrameCodec
                └── gRPC 핸들러들

// 채널을 다 쓴 후에는 반드시 종료
// 내부적으로 Netty의 EventLoopGroup.shutdownGracefully() 호출
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);

Java 객체 (HelloRequest)
  ↓ protobuf 직렬화
바이트 배열 (Protocol Buffers 인코딩)
  ↓ gRPC 프레이밍
gRPC Length-Prefixed Message
  ↓ HTTP/2 인코딩
HTTP/2 HEADERS 프레임 + DATA 프레임
  ↓ Netty → TCP
네트워크 전송

┌──────────────┬────────────────────────┐
│ Compressed   │ Message Length          │
│ Flag (1byte) │ (4 bytes, big-endian)  │
├──────────────┴────────────────────────┤
│ Serialized Protocol Buffers Message   │
│ (가변 길이)                            │
└───────────────────────────────────────┘

요청 시:
  HEADERS 프레임
  ┌─ :method = POST
  ├─ :path = /GreeterService/SayHello
  ├─ :scheme = http
  ├─ content-type = application/grpc
  ├─ te = trailers
  └─ grpc-encoding = identity

  DATA 프레임
  └─ [5바이트 gRPC 헤더 + protobuf 바이트]

응답 시:
  HEADERS 프레임
  ├─ :status = 200
  └─ content-type = application/grpc

  DATA 프레임
  └─ [5바이트 gRPC 헤더 + protobuf 바이트]

  HEADERS 프레임 (trailers)
  ├─ grpc-status = 0
  └─ grpc-message = (비어있음 = 성공)

HTTP/1.1 — 연결당 하나의 요청
┌──────────────────────────────────────┐
│ Connection 1: RPC-A 요청 → 응답 대기  │
│ Connection 2: RPC-B 요청 → 응답 대기  │
│ Connection 3: RPC-C 요청 → 응답 대기  │
│ (연결 3개 필요)                       │
└──────────────────────────────────────┘

gRPC (HTTP/2) — 하나의 연결에서 모든 요청
┌──────────────────────────────────────┐
│ Connection 1:                        │
│   Stream 1: RPC-A ─────────────────  │
│   Stream 3: RPC-B ──────────         │
│   Stream 5: RPC-C ────              │
│ (연결 1개로 충분)                     │
└──────────────────────────────────────┘

Netty Channel (TCP 연결)
├── Http2FrameCodec        ← HTTP/2 프레임 인코딩/디코딩
├── Http2MultiplexHandler  ← 스트림을 자식 Channel로 분리
│
├── Http2StreamChannel (Stream 1) ← RPC-A
│   └── gRPC 스트림 핸들러
├── Http2StreamChannel (Stream 3) ← RPC-B
│   └── gRPC 스트림 핸들러
└── Http2StreamChannel (Stream 5) ← RPC-C
    └── gRPC 스트림 핸들러

// gRPC 클라이언트 인터셉터 — 모든 요청에 인증 토큰 추가
public class AuthInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
            MethodDescriptor<ReqT, RespT> method,
            CallOptions callOptions,
            Channel next) {

        return new ForwardingClientCall
                .SimpleForwardingClientCall<>(next.newCall(method, callOptions)) {

            @Override
            public void start(Listener<RespT> listener, Metadata headers) {
                // 메타데이터에 인증 토큰 추가
                headers.put(
                    Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER),
                    "Bearer my-token"
                );
                super.start(listener, headers);
            }
        };
    }
}

// 인터셉터를 채널에 등록
ManagedChannel channel = ManagedChannelBuilder
    .forAddress("localhost", 50051)
    .intercept(new AuthInterceptor())  // 체인에 추가
    .build();

// gRPC 서버 인터셉터 — 요청 처리 시간 로깅
public class LoggingInterceptor implements ServerInterceptor {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {

        long startTime = System.nanoTime();
        String methodName = call.getMethodDescriptor().getFullMethodName();

        // 다음 핸들러(또는 실제 서비스)로 전달
        ServerCall.Listener<ReqT> listener = next.startCall(call, headers);

        return new ForwardingServerCallListener
                .SimpleForwardingServerCallListener<>(listener) {

            @Override
            public void onComplete() {
                long elapsed = System.nanoTime() - startTime;
                // 처리 시간 로깅
                System.out.printf("%s 완료: %dms%n",
                    methodName, elapsed / 1_000_000);
                super.onComplete();
            }
        };
    }
}

gRPC 인터셉터 체인:
  클라이언트 코드
    → AuthInterceptor
    → LoggingInterceptor
    → 실제 RPC 전송 (여기서 Netty를 사용)

Netty ChannelPipeline:
  네트워크 바이트
    → SslHandler
    → Http2FrameCodec
    → gRPC 내부 핸들러
    → 비즈니스 로직

import io.grpc.netty.NettyChannelBuilder;

ManagedChannel channel = NettyChannelBuilder
    .forAddress("localhost", 50051)
    .usePlaintext()
    .build();

// EventLoopGroup을 직접 생성하여 여러 채널이 공유
EventLoopGroup sharedGroup = new NioEventLoopGroup(4);

ManagedChannel channelA = NettyChannelBuilder
    .forAddress("service-a", 50051)
    .eventLoopGroup(sharedGroup)
    .channelType(NioSocketChannel.class)
    .usePlaintext()
    .build();

ManagedChannel channelB = NettyChannelBuilder
    .forAddress("service-b", 50052)
    .eventLoopGroup(sharedGroup)
    .channelType(NioSocketChannel.class)
    .usePlaintext()
    .build();

// 주의: 공유 EventLoopGroup은 채널 종료 시 자동으로 닫히지 않음
// 반드시 직접 종료해야 함
channelA.shutdown();
channelB.shutdown();
sharedGroup.shutdownGracefully();

ManagedChannel channel = NettyChannelBuilder
    .forAddress("localhost", 50051)
    // keepAlive 핑 간격 (기본값 없음 — 비활성)
    .keepAliveTime(30, TimeUnit.SECONDS)
    // keepAlive 핑 응답 대기 시간 (기본 20초)
    .keepAliveTimeout(10, TimeUnit.SECONDS)
    // 활성 RPC가 없어도 keepAlive 핑을 보낼지 (기본 false)
    .keepAliveWithoutCalls(true)
    .build();

import io.grpc.netty.NettyServerBuilder;

// Netty 기반 gRPC 서버 설정
Server server = NettyServerBuilder
    .forPort(50051)
    // 워커 EventLoopGroup 스레드 수 지정
    .workerEventLoopGroup(new NioEventLoopGroup(8))
    .channelType(NioServerSocketChannel.class)
    // 최대 동시 스트림 수 (HTTP/2 SETTINGS)
    .maxConcurrentCallsPerConnection(100)
    // 수신 메시지 최대 크기 (기본 4MB)
    .maxInboundMessageSize(16 * 1024 * 1024)
    // 연결 유지 설정
    .keepAliveTime(2, TimeUnit.HOURS)
    .keepAliveTimeout(20, TimeUnit.SECONDS)
    // 클라이언트의 최소 keepAlive 간격 허용치
    .permitKeepAliveTime(10, TimeUnit.SECONDS)
    .permitKeepAliveWithoutCalls(true)
    .addService(new GreeterServiceImpl())
    .build()
    .start();

1. stub.sayHello(request)
   └── gRPC 스텁이 RPC 호출 시작

2. 인터셉터 체인 통과
   └── AuthInterceptor → LoggingInterceptor → ...

3. ManagedChannel → Subchannel 선택
   └── LoadBalancer가 적절한 Subchannel(서버) 선택

4. Protocol Buffers 직렬화
   └── HelloRequest → 바이트 배열

5. gRPC 프레이밍
   └── 5바이트 헤더(compressed flag + length) + protobuf 바이트

6. HTTP/2 인코딩 (Netty ChannelPipeline)
   └── HEADERS 프레임 + DATA 프레임 생성

7. TLS 암호화 (SslHandler)
   └── HTTP/2 프레임을 암호화

8. TCP 전송
   └── Netty의 EventLoop가 소켓에 쓰기