Microservice Patterns

Microservice Patterns — 微服务设计模式

微服务架构的核心设计模式和最佳实践。

Goal

微服务架构设计模式，包含服务拆分、通信方式、数据管理、服务发现、熔断降级、链路追踪

Trigger

• 用户要求"微服务架构"、"服务拆分"
  • 需要设计分布式系统
  • 需要实现服务治理

Workflow

1. 分析业务域 — DDD 限界上下文划分
2. 设计服务边界 — 单一职责、高内聚低耦合
3. 选择通信方式 — 同步 REST/gRPC / 异步消息
4. 设计数据策略 — 数据库 per 服务 / 共享数据库
5. 实现治理 — 服务发现、熔断、限流

服务拆分原则

拆分策略

拆分反模式

通信模式

同步通信 (HTTP/gRPC)

服务A ──HTTP/gRPC──▶ 服务B
       ◀──Response──

异步通信 (消息队列)

服务A ──Publish──▶ 消息队列 ──Consume──▶ 服务B

同步实现示例

# gRPC 服务定义
# user.proto
syntax = "proto3";

service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
}

message GetUserRequest {
  string id = 1;
}

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

# gRPC 客户端
import grpc
from user_pb2_grpc import UserServiceStub
from user_pb2 import GetUserRequest

class UserClient:
    def __init__(self, host: str):
        self.channel = grpc.insecure_channel(host)
        self.stub = UserServiceStub(self.channel)
    
    def get_user(self, user_id: str):
        request = GetUserRequest(id=user_id)
        return self.stub.GetUser(request)

数据管理

每服务一数据库

用户服务 ──▶ 用户数据库 (PostgreSQL)
订单服务 ──▶ 订单数据库 (MySQL)
商品服务 ──▶ 商品数据库 (MongoDB)

Saga 模式 (分布式事务)

创建订单流程:
1. 订单服务: 创建订单 (PENDING)
2. 库存服务: 预扣库存
3. 支付服务: 扣款
4. 订单服务: 更新订单状态 (PAID)

补偿流程 (任一步骤失败):
3. 支付服务: 退款
2. 库存服务: 恢复库存
1. 订单服务: 取消订单

# Saga 编排器
class OrderSaga:
    def __init__(self):
        self.steps = [
            SagaStep(
                action=self.create_order,
                compensation=self.cancel_order
            ),
            SagaStep(
                action=self.reserve_inventory,
                compensation=self.release_inventory
            ),
            SagaStep(
                action=self.process_payment,
                compensation=self.refund_payment
            ),
        ]
    
    async def execute(self, order_data):
        completed = []
        
        for step in self.steps:
            try:
                result = await step.action(order_data)
                completed.append(step)
            except Exception as e:
                # 补偿已完成的步骤
                for completed_step in reversed(completed):
                    await completed_step.compensation(order_data)
                raise SagaFailedError(str(e))

服务发现

注册中心

客户端发现

# Consul 服务发现
import consul

class ServiceDiscovery:
    def __init__(self):
        self.consul = consul.Consul()
    
    def register(self, name: str, host: str, port: int):
        self.consul.agent.service.register(
            name=name,
            service_id=f"{name}-{host}-{port}",
            address=host,
            port=port,
            check=consul.Check.http(f"http://{host}:{port}/health", interval="10s")
        )
    
    def discover(self, name: str) -> tuple:
        _, services = self.consul.health.service(name, passing=True)
        if not services:
            raise ServiceNotFoundError(name)
        service = random.choice(services)
        return service['Service']['Address'], service['Service']['Port']

熔断降级

熔断器模式

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"        # 正常
    OPEN = "open"            # 熔断
    HALF_OPEN = "half_open"  # 半开

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitOpenError("Circuit is open")
        
        try:
            result = await func(*args, **kwargs)
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            
            raise

链路追踪

OpenTelemetry 集成

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 初始化
provider = TracerProvider()
jaeger_exporter = JaegerExporter(agent_host_name="localhost", agent_port=6831)
provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# 使用
async def create_order(order_data):
    with tracer.start_as_current_span("create_order") as span:
        span.set_attribute("order.amount", order_data["amount"])
        
        # 调用其他服务
        with tracer.start_as_current_span("call_inventory_service"):
            await inventory_service.reserve(order_data["items"])
        
        with tracer.start_as_current_span("call_payment_service"):
            await payment_service.charge(order_data["amount"])

Example

# 设计微服务架构
为电商平台设计微服务架构

# 实现服务间通信
实现 gRPC 服务间调用

# 实现分布式事务
使用 Saga 模式实现订单创建流程

# 添加熔断降级
为服务调用添加熔断器

# 配置链路追踪
集成 OpenTelemetry 链路追踪

参考资料

• 微服务模式: references/patterns.md
• Saga 模式: references/saga.md