系统设计与通信实战指南：从CDN到服务网格

# 静态资源缓存
location ~* \.(jpg|png|css|js|woff2)$ {
    expires 1y;
    add_header Cache-Control "public, immutable";
}

# HTML 不缓存
location / {
    add_header Cache-Control "no-cache, no-store, must-revalidate";
}

# CORS 配置
add_header Access-Control-Allow-Origin "https://example.com";
add_header Access-Control-Allow-Methods "GET, OPTIONS";

// user.proto
syntax = "proto3";

package user;

service UserService {
    rpc GetUser (GetUserRequest) returns (User);
    rpc ListUsers (ListUsersRequest) returns (stream User);
    rpc CreateUsers (stream CreateUserRequest) returns (CreateUsersResponse);
    rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

message GetUserRequest {
    int32 id = 1;
}

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

# 服务端
import grpc
from concurrent import futures

class UserService(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = db.get_user(request.id)
        return user_pb2.User(
            id=user.id,
            name=user.name,
            email=user.email
        )

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

# 客户端
channel = grpc.insecure_channel('localhost:50051')
stub = user_pb2_grpc.UserServiceStub(channel)
response = stub.GetUser(user_pb2.GetUserRequest(id=1))
print(response.name)

# 服务端拦截器（鉴权、日志）
class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = handler_call_details.invocation_metadata
        token = next((v for k, v in metadata if k == 'authorization'), None)

        if not self._verify_token(token):
            return self._unary_unary_rpc_terminator

        return continuation(handler_call_details)

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

# 服务端流式（适合大数据量）
class UserService:
    def ListUsers(self, request, context):
        for user in db.get_all_users():
            yield user_pb2.User(id=user.id, name=user.name)

# 双向流式（适合聊天）
class ChatService:
    def Chat(self, request_iterator, context):
        for message in request_iterator:
            response = process_message(message)
            yield response

type User {
    id: ID!
    name: String!
    email: String
    posts: [Post!]!
}

type Post {
    id: ID!
    title: String!
    content: String!
    author: User!
}

type Query {
    user(id: ID!): User
    users(limit: Int = 10): [User!]!
}

type Mutation {
    createUser(name: String!, email: String!): User!
    updateUser(id: ID!, name: String): User!
}

type Subscription {
    postCreated: Post!
}

import strawberry

@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> User:
        return db.get_user(id)

    @strawberry.field
    def users(self, limit: int = 10) -> list[User]:
        return db.get_users(limit)

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str, email: str) -> User:
        return db.create_user(name, email)

schema = strawberry.Schema(query=Query, mutation=Mutation)

# 错误：N+1 查询
def resolve_posts(user):
    return [db.get_post(pid) for pid in user.post_ids]  # N 次查询

# 正确：用 DataLoader 批量查询
from strawberry.dataloader import DataLoader

async def load_posts(post_ids):
    posts = db.get_posts_batch(post_ids)
    return [posts.get(pid) for pid in post_ids]

post_loader = DataLoader(load_posts)

def resolve_posts(user):
    return [post_loader.load(pid) for pid in user.post_ids]

from graphql import validate
from graphql.depth_limit import depth_limit_validator

schema = build_schema(...)
validate(query, schema, [depth_limit_validator(max_depth=10)])

from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class Event:
    aggregate_id: str
    event_type: str
    data: dict
    timestamp: datetime

class EventStore:
    def __init__(self):
        self.events = {}  # 实际用数据库

    def append(self, aggregate_id: str, event: Event):
        if aggregate_id not in self.events:
            self.events[aggregate_id] = []
        self.events[aggregate_id].append(event)

    def get_events(self, aggregate_id: str) -> List[Event]:
        return self.events.get(aggregate_id, [])

# 订单聚合
class OrderAggregate:
    def __init__(self):
        self.id = None
        self.status = None
        self.items = []

    def apply_event(self, event: Event):
        if event.event_type == 'OrderCreated':
            self.id = event.data['order_id']
            self.items = event.data['items']
            self.status = 'created'
        elif event.event_type == 'OrderPaid':
            self.status = 'paid'
        elif event.event_type == 'OrderShipped':
            self.status = 'shipped'

    @classmethod
    def from_events(cls, events: List[Event]):
        order = cls()
        for event in events:
            order.apply_event(event)
        return order

# 使用
store = EventStore()

# 创建订单（产生事件）
store.append('order-123', Event(
    'order-123', 'OrderCreated',
    {'order_id': 'order-123', 'items': ['book', 'pen']},
    datetime.now()
))

# 支付
store.append('order-123', Event(
    'order-123', 'OrderPaid', {}, datetime.now()
))

# 重建订单状态
events = store.get_events('order-123')
order = OrderAggregate.from_events(events)
print(order.status)  # 'paid'

# 写模型：处理命令
class OrderCommandHandler:
    def handle_create_order(self, command):
        # 验证、产生事件
        event = Event('OrderCreated', {...})
        event_store.append(command.order_id, event)
        # 发布事件到消息队列
        message_bus.publish(event)

# 读模型：优化查询
class OrderReadModel:
    """从事件构建的物化视图"""
    def __init__(self):
        self.orders = {}  # 实际用数据库/ES

    def handle_event(self, event):
        if event.event_type == 'OrderCreated':
            self.orders[event.aggregate_id] = event.data
        elif event.event_type == 'OrderPaid':
            self.orders[event.aggregate_id]['status'] = 'paid'

    def get_order(self, order_id):
        return self.orders.get(order_id)

# 1. 超时
import httpx

async def call_service():
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.post('http://service-b/api', json=data)
            return response.json()
        except httpx.TimeoutException:
            # 超时降级
            return fallback()

# 2. 重试（指数退避）
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def call_with_retry():
    return await call_service()

# 3. 熔断
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
async def call_with_circuit():
    return await call_service()

# 4. 幂等
async def create_order(order_id, data):
    # 幂等检查
    if order_exists(order_id):
        return get_order(order_id)
    return await create_new_order(order_id, data)

# 1. 流量管理
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: myapp
spec:
  http:
  - match:
    - headers:
        canary:
          exact: "true"
    route:
    - destination:
        host: myapp
        subset: v2  # 金丝雀版本
  - route:
    - destination:
        host: myapp
        subset: v1  # 默认版本
      weight: 90
    - destination:
        host: myapp
        subset: v2
      weight: 10  # 10% 流量到 v2

# 2. 熔断
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: myapp
spec:
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 30s

# 3. mTLS
apiVersion: security.istio.io/v1alpha3
kind: PeerAuthentication
metadata:
  name: default
spec:
  mtls:
    mode: STRICT

# 混沌实验模板
experiment = {
    "name": "验证订单服务宕机时系统是否降级",
    "hypothesis": "订单服务宕机时，购物车服务应降级为'稍后重试'",
    "steady_state": {
        "metric": "error_rate",
        "threshold": "< 1%"
    },
    "failure": {
        "type": "service_down",
        "target": "order-service",
        "duration": "5m"
    },
    "rollback": {
        "condition": "error_rate > 5%",
        "action": "restore order-service"
    }
}

# Pod 故障注入
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-kill-example
spec:
  action: pod-kill
  mode: one
  selector:
    namespaces:
      - default
    labelSelectors:
      app: myapp
  scheduler:
    cron: "@every 10m"

# 网络延迟
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-delay
spec:
  action: delay
  mode: all
  selector:
    labelSelectors:
      app: myapp
  delay:
    latency: "100ms"
    correlation: "100"
    jitter: "10ms"
  duration: "30s"