Split View: gRPC 심화 완전 가이드 2025: Protocol Buffers, 4가지 통신 모델, 인터셉터, 로드 밸런싱

GET /api/users/123 HTTP/1.1
Host: api.example.com

HTTP/1.1 200 OK
Content-Type: application/json
{"id": 123, "name": "Alice", "email": "alice@example.com"}

service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc StreamUsers(Empty) returns (stream User);
}

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
}

syntax = "proto3";

package myapp;

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  repeated string tags = 4;
  Address address = 5;
}

message Address {
  string street = 1;
  string city = 2;
}

message User {
  int64 id = 1;      // ← 1은 필드 번호
  string name = 2;
}

// v1
message User {
  string user_name = 1;
}

// v2 (호환됨)
message User {
  string display_name = 1;  // 이름 바꿈, 번호 같음
}

┌─────────────┬──────────────┐
│ field_number│  field_value │
│   + type    │              │
└─────────────┴──────────────┘

0      → 1 byte
127    → 1 byte
128    → 2 bytes
16383  → 2 bytes
16384  → 3 bytes

// JSON: 78 bytes
{"id":123,"name":"Alice","email":"alice@example.com"}

// Protobuf: 35 bytes (45% 작음)
0a 03 31 32 33  # field 1 (id), len 3, "123"
12 05 41 6c 69 63 65  # field 2 (name), len 5, "Alice"
1a 11 ...

message User {
  reserved 4, 5;          // 4, 5는 사용 안 함
  reserved "old_field";    // 이름도 reserved
  
  int64 id = 1;
  string name = 2;
  string email = 3;
}

rpc GetUser(GetUserRequest) returns (User);

# Client
response = stub.GetUser(GetUserRequest(user_id=123))
print(response.name)

# Server
def GetUser(self, request, context):
    user = db.get(request.user_id)
    return User(id=user.id, name=user.name, email=user.email)

rpc StreamUsers(Empty) returns (stream User);

# Client
for user in stub.StreamUsers(Empty()):
    print(user.name)

# Server
def StreamUsers(self, request, context):
    for user in db.iter_all():
        yield User(id=user.id, name=user.name)

rpc UploadEvents(stream Event) returns (UploadResponse);

# Client
def event_generator():
    for event in events:
        yield event

response = stub.UploadEvents(event_generator())
print(response.processed)

# Server
def UploadEvents(self, request_iterator, context):
    count = 0
    for event in request_iterator:
        process(event)
        count += 1
    return UploadResponse(processed=count)

rpc Chat(stream ChatMessage) returns (stream ChatMessage);

# Client
def send_messages():
    yield ChatMessage(text="Hello")
    yield ChatMessage(text="How are you?")

for response in stub.Chat(send_messages()):
    print(f"Received: {response.text}")

# Server
def Chat(self, request_iterator, context):
    for msg in request_iterator:
        yield ChatMessage(text=f"Echo: {msg.text}")

[Client] ─────────→ [Server]
   ↓
[Connection 1: Request 1]
[Connection 2: Request 2]
[Connection 3: Request 3]
   ...

[Client] ─────────→ [Server]
   ↓
[Single Connection]
   ├─ Stream 1: Request 1 ↔ Response 1
   ├─ Stream 2: Request 2 ↔ Response 2
   └─ Stream 3: Request 3 ↔ Response 3

GET /api/users/123 HTTP/1.1
Host: api.example.com
User-Agent: MyApp/1.0
Accept: application/json
Authorization: Bearer xyz...

import grpc

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        token = metadata.get('authorization')
        
        if not verify_token(token):
            return grpc.unary_unary_rpc_method_handler(
                lambda req, ctx: ctx.abort(grpc.StatusCode.UNAUTHENTICATED, 'Invalid token')
            )
        
        return continuation(handler_call_details)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[AuthInterceptor()]
)

class RetryInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        for attempt in range(3):
            try:
                return continuation(client_call_details, request)
            except grpc.RpcError as e:
                if e.code() != grpc.StatusCode.UNAVAILABLE:
                    raise
                time.sleep(2 ** attempt)
        raise

channel = grpc.intercept_channel(
    grpc.insecure_channel('localhost:50051'),
    RetryInterceptor()
)

func loggingInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    start := time.Now()
    resp, err := handler(ctx, req)
    log.Printf("%s took %v", info.FullMethod, time.Since(start))
    return resp, err
}

server := grpc.NewServer(
    grpc.UnaryInterceptor(loggingInterceptor),
)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[
        TracingInterceptor(),
        AuthInterceptor(),
        LoggingInterceptor(),
        MetricsInterceptor(),
    ]
)

# 잘못 - 무한 대기
response = stub.GetUser(GetUserRequest(user_id=123))

# 올바름 - 5초 deadline
response = stub.GetUser(GetUserRequest(user_id=123), timeout=5.0)

[Client] (timeout=5s)
   ↓ deadline=now+5s
[Service A] (4s 소요)
   ↓ remaining=1s
[Service B] (이미 1s 남았으므로 짧게)

def call_chain(context):
    # context의 deadline 자동 사용
    response = downstream_stub.SomeRPC(request, timeout=context.time_remaining())

import grpc

# 클라이언트가 취소
future = stub.GetUser.future(request)
time.sleep(2)
future.cancel()  # 서버에 취소 신호 전파

def GetUser(self, request, context):
    while not context.is_active():  # 클라이언트가 취소했나?
        return None
    
    # 또는
    if context.cancelled():
        return None

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

response, err := client.GetUser(ctx, &GetUserRequest{UserId: 123})

# DNS-based
channel = grpc.insecure_channel(
    'dns:///my-service:50051',
    options=[
        ('grpc.lb_policy_name', 'round_robin')
    ]
)

[Client] → [LB Service] → "사용 가능한 서버: A, B, C"
   ↓
[Client] → [Server A]

[Client] → [Envoy Proxy] → [Servers]

syntax = "proto3";

package grpc.health.v1;

service Health {
  rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
  rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
}

message HealthCheckResponse {
  enum ServingStatus {
    UNKNOWN = 0;
    SERVING = 1;
    NOT_SERVING = 2;
  }
  ServingStatus status = 1;
}

options = [
    ('grpc.max_send_message_length', 100 * 1024 * 1024),  # 100MB
    ('grpc.max_receive_message_length', 100 * 1024 * 1024),
]

channel = grpc.insecure_channel('localhost:50051', options=options)

# 잘못 - 매 요청마다 새 연결
def get_user(user_id):
    channel = grpc.insecure_channel('localhost:50051')
    stub = UserServiceStub(channel)
    return stub.GetUser(GetUserRequest(user_id=user_id))

# 올바름 - 채널 재사용
channel = grpc.insecure_channel('localhost:50051')
stub = UserServiceStub(channel)

def get_user(user_id):
    return stub.GetUser(GetUserRequest(user_id=user_id))

options = [
    ('grpc.keepalive_time_ms', 10000),       # 10초마다 ping
    ('grpc.keepalive_timeout_ms', 5000),     # ping 응답 5초 대기
    ('grpc.keepalive_permit_without_calls', True),
    ('grpc.http2.max_pings_without_data', 0),
]

# Server
server = grpc.server(..., compression=grpc.Compression.Gzip)

# Client
channel = grpc.insecure_channel(
    'localhost:50051',
    compression=grpc.Compression.Gzip
)

[Browser] ─ gRPC-Web ─→ [Envoy Proxy] ─ gRPC ─→ [Server]

import { UserServiceClient } from './generated/user_pb_service'

const client = new UserServiceClient('https://api.example.com')

client.getUser(new GetUserRequest().setUserId(123), (err, response) => {
    if (err) console.error(err)
    else console.log(response.getName())
})

import { createPromiseClient } from "@bufbuild/connect"
import { UserService } from "./gen/user_connect"

const client = createPromiseClient(UserService, transport)
const response = await client.getUser({ userId: 123 })

# 서비스 목록
grpcurl -plaintext localhost:50051 list

# 메서드 호출
grpcurl -plaintext -d '{"user_id": 123}' \
  localhost:50051 \
  UserService/GetUser

from grpc_reflection.v1alpha import reflection
SERVICE_NAMES = (UserService_pb2.DESCRIPTOR.services_by_name['UserService'].full_name, reflection.SERVICE_NAME)
reflection.enable_server_reflection(SERVICE_NAMES, server)

import grpc
import logging

logging.basicConfig(level=logging.DEBUG)
os.environ['GRPC_VERBOSITY'] = 'DEBUG'
os.environ['GRPC_TRACE'] = 'all'

from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer

GrpcInstrumentorServer().instrument()