✍️ 필사 모드: 마이크로서비스 통신 패턴 완전 가이드 2025: REST vs gRPC vs 비동기 메시징, Saga, Outbox

GET /api/v1/users/1000 HTTP/1.1
Host: api.example.com
Accept: application/json
Authorization: Bearer xyz...

HTTP/1.1 200 OK
Content-Type: application/json

{
  "id": 1000,
  "name": "Alice",
  "email": "alice@example.com"
}

// user.proto
syntax = "proto3";

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

message GetUserRequest {
  int64 user_id = 1;
}

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

query GetUserWithPosts {
  user(id: 1000) {
    name
    email
    posts(limit: 10) {
      title
      createdAt
    }
  }
}

order-events       # 주문 도메인 이벤트
payment-events     # 결제 도메인 이벤트
inventory-events   # 재고 도메인 이벤트

producer.send(
    'order-events',
    key=str(order_id).encode(),  # 같은 키 → 같은 파티션
    value=json.dumps(event).encode()
)

[Order Service]
   ↓ "OrderCreated" event
[Payment Service]
   ↓ "PaymentProcessed" event
[Inventory Service]
   ↓ "InventoryReserved" event
[Shipping Service]

[Saga Orchestrator]
   ├─ Call Payment Service
   ├─ Call Inventory Service
   └─ Call Shipping Service
   (실패 시 보상 호출)

@workflow.defn
class OrderSaga:
    @workflow.run
    async def run(self, order_id: str):
        try:
            await workflow.execute_activity(
                charge_payment, order_id,
                start_to_close_timeout=timedelta(seconds=30)
            )
            await workflow.execute_activity(
                reserve_inventory, order_id,
                start_to_close_timeout=timedelta(seconds=30)
            )
            await workflow.execute_activity(
                create_shipment, order_id,
                start_to_close_timeout=timedelta(seconds=30)
            )
        except ActivityError:
            # 자동 보상
            await workflow.execute_activity(refund_payment, order_id)
            raise

def create_order(order):
    db.save(order)              # 1. DB 저장
    kafka.send("order-events", order)  # 2. Kafka 발행

CREATE TABLE outbox (
  id UUID PRIMARY KEY,
  aggregate_type VARCHAR(255),
  aggregate_id VARCHAR(255),
  event_type VARCHAR(255),
  payload JSONB,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  processed_at TIMESTAMPTZ
);

def create_order(order):
    with db.transaction():
        db.save(order)
        db.save(OutboxEvent(
            aggregate_type='Order',
            aggregate_id=order.id,
            event_type='OrderCreated',
            payload=order.to_dict()
        ))
    # 트랜잭션 commit 시 두 INSERT가 원자적

while True:
    events = db.query("SELECT * FROM outbox WHERE processed_at IS NULL LIMIT 100")
    for event in events:
        kafka.send(event.aggregate_type, event.payload)
        db.update("UPDATE outbox SET processed_at = NOW() WHERE id = ?", event.id)
    sleep(1)

# Debezium connector config
{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "orders",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter"
  }
}

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def call_payment_service(order):
    response = requests.post('http://payment/charge', json=order)
    return response.json()

import time
import random

def retry_with_backoff(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)  # 지수 + jitter
            time.sleep(wait)

# 서비스마다 별도 스레드 풀
payment_pool = ThreadPoolExecutor(max_workers=10)
inventory_pool = ThreadPoolExecutor(max_workers=10)

requests.get(url, timeout=(3, 10))  # connect 3s, read 10s

def process_payment(idempotency_key, amount):
    if cache.exists(f"payment:{idempotency_key}"):
        return cache.get(f"payment:{idempotency_key}")  # 이전 결과 반환
    
    result = charge_card(amount)
    cache.set(f"payment:{idempotency_key}", result, ttl=86400)
    return result

POST /v1/charges HTTP/1.1
Idempotency-Key: my-unique-key-123

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def process_order(order):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order.id)
        with tracer.start_as_current_span("charge_payment"):
            payment_service.charge(order)
        with tracer.start_as_current_span("reserve_inventory"):
            inventory_service.reserve(order)

GET /api/inventory HTTP/1.1
traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01