3.4.1 微服务拆分 · 4 个原则(单一职责/业务边界/数据独立/团队匹配)

# ❌ 反例:OrderService 什么都做
class OrderService:
    def create_order(self): ...
    def cancel_order(self): ...
    def query_user(self): ...        # 应该是 UserService
    def deduct_inventory(self): ...  # 应该是 InventoryService
    def charge_payment(self): ...    # 应该是 PaymentService
    def ship_logistics(self): ...    # 应该是 LogisticsService
    def generate_report(self): ...   # 应该是 ReportService
    def send_sms(self): ...          # 应该是 NotificationService

# user_service/app.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, EmailStr
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

app = FastAPI(title="user-service")
engine = create_engine("mysql+pymysql://user:pass@mysql/users")
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String(120), unique=True, nullable=False)
    name = Column(String(80), nullable=False)
    status = Column(String(20), default="active")


class UserCreate(BaseModel):
    email: EmailStr
    name: str


class UserOut(BaseModel):
    id: int
    email: EmailStr
    name: str

    class Config:
        from_attributes = True


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users", response_model=UserOut, status_code=201)
def create_user(payload: UserCreate, db=Depends(get_db)):
    user = User(email=payload.email, name=payload.name)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int, db=Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="user not found")
    return user


@app.patch("/users/{user_id}/status")
def deactivate(user_id: int, db=Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(404)
    user.status = "inactive"
    db.commit()
    return {"id": user_id, "status": "inactive"}

# order_service/client/inventory_client.py
import httpx

class InventoryClient:
    def __init__(self, base_url: str = "http://inventory-svc"):
        self.base_url = base_url

    def reserve(self, sku: str, qty: int) -> dict:
        r = httpx.post(
            f"{self.base_url}/reservations",
            json={"sku": sku, "qty": qty},
            timeout=2.0,
        )
        r.raise_for_status()
        return r.json()

    def release(self, reservation_id: int) -> None:
        httpx.delete(
            f"{self.base_url}/reservations/{reservation_id}",
            timeout=2.0,
        ).raise_for_status()

# shared/events.py
from pydantic import BaseModel
from datetime import datetime

class DomainEvent(BaseModel):
    event_id: str
    occurred_at: datetime
    event_type: str


class OrderCreated(DomainEvent):
    order_id: int
    user_id: int
    items: list[dict]  # [{"sku": "A1", "qty": 2, "price": 99}]

    event_type: str = "order.created"

# order_service/publisher.py
import json, aio_pika, uuid
from datetime import datetime
from shared.events import OrderCreated

async def publish_order_created(order: dict):
    connection = await aio_pika.connect_robust("amqp://guest:guest@rabbitmq/")
    channel = await connection.channel()
    await channel.default_exchange.publish(
        aio_pika.Message(
            body=OrderCreated(
                event_id=str(uuid.uuid4()),
                occurred_at=datetime.utcnow(),
                order_id=order["id"],
                user_id=order["user_id"],
                items=order["items"],
            ).model_dump_json().encode(),
        ),
        routing_key="order.created",
    )
    await connection.close()

# inventory_service/consumer.py
import aio_pika, json
from inventory_service.db import reserve_stock

async def consume_order_created():
    connection = await aio_pika.connect_robust("amqp://guest:guest@rabbitmq/")
    channel = await connection.channel()
    queue = await channel.declare_queue("inventory.order_created", durable=True)
    await queue.bind(channel.default_exchange, routing_key="order.created")

    async with queue.iterator() as it:
        async for msg in it:
            async with msg.process():
                payload = json.loads(msg.body)
                await reserve_stock(
                    order_id=payload["order_id"],
                    items=payload["items"],
                )

# saga/order_saga.py
from dataclasses import dataclass
from enum import Enum

class SagaState(str, Enum):
    PENDING = "pending"
    INVENTORY_RESERVED = "inventory_reserved"
    PAYMENT_CHARGED = "payment_charged"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"


@dataclass
class OrderSaga:
    saga_id: str
    order_id: int
    state: SagaState = SagaState.PENDING
    inventory_reservation_id: int | None = None
    payment_txn_id: str | None = None

    async def step_1_reserve_inventory(self, inv_client):
        """正向操作:预扣库存"""
        result = await inv_client.reserve(self.sku, self.qty)
        self.inventory_reservation_id = result["reservation_id"]
        self.state = SagaState.INVENTORY_RESERVED

    async def step_2_charge_payment(self, pay_client):
        """正向操作:支付收款"""
        result = await pay_client.charge(self.amount)
        self.payment_txn_id = result["txn_id"]
        self.state = SagaState.PAYMENT_CHARGED

    async def compensate(self, inv_client, pay_client):
        """补偿操作:释放库存 + 退款"""
        self.state = SagaState.COMPENSATING
        if self.inventory_reservation_id:
            await inv_client.release(self.inventory_reservation_id)
        if self.payment_txn_id:
            await pay_client.refund(self.payment_txn_id)
        self.state = SagaState.FAILED

# ❌ 反例:OrderService 直接读 InventoryService 的表
def create_order_bad(sku: str, qty: int, db_session):
    # 直接 JOIN 库存表
    stock = db_session.execute(
        "SELECT qty FROM inventory_db.stock WHERE sku = :sku FOR UPDATE",
        {"sku": sku},
    ).fetchone()
    if stock.qty < qty:
        raise OutOfStock()
    # ...

# bff/order_detail.py
# 聚合服务(Backend For Frontend)串行调用多个微服务
import asyncio
import httpx

async def get_order_detail(order_id: int) -> dict:
    async with httpx.AsyncClient(timeout=2.0) as client:
        order, payment, logistics = await asyncio.gather(
            client.get(f"http://order-svc/orders/{order_id}"),
            client.get(f"http://payment-svc/payments?order_id={order_id}"),
            client.get(f"http://logistics-svc/shipments?order_id={order_id}"),
        )
    return {
        "order": order.json(),
        "payment": payment.json(),
        "logistics": logistics.json(),
    }

# query_service/models.py
# 查询服务维护一个宽表,定时从各服务同步
from sqlalchemy import Column, Integer, String, Numeric, DateTime
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class OrderDetailView(Base):
    """宽表:订单 + 支付 + 物流 的去规范化物化视图"""
    __tablename__ = "order_detail_view"

    order_id = Column(Integer, primary_key=True)
    user_id = Column(Integer)
    sku = Column(String(50))
    amount = Column(Numeric(10, 2))
    payment_status = Column(String(20))
    logistics_status = Column(String(20))
    updated_at = Column(DateTime)

# query_service/sync.py
# 订阅 Domain Event,更新宽表
import json, aio_pika
from query_service.db import SessionLocal
from query_service.models import OrderDetailView

async def on_order_created(payload: dict):
    db = SessionLocal()
    row = OrderDetailView(
        order_id=payload["order_id"],
        user_id=payload["user_id"],
        sku=payload["items"][0]["sku"],
        amount=sum(i["price"] * i["qty"] for i in payload["items"]),
        payment_status="pending",
        logistics_status="pending",
        updated_at=payload["occurred_at"],
    )
    db.merge(row)
    db.commit()


async def on_payment_charged(payload: dict):
    db = SessionLocal()
    row = db.query(OrderDetailView).filter_by(
        order_id=payload["order_id"]
    ).first()
    row.payment_status = payload["status"]
    row.updated_at = payload["occurred_at"]
    db.commit()

# binlog relay(伪代码:用 Debezium 监听 MySQL binlog)
# kafka topic: mysql.inventory.stock

# inventory_service/projector.py
# 把库存变动投影到 Redis(读模型)
import json, redis
from aiokafka import AIOKafkaConsumer

r = redis.Redis(host="redis", port=6379, decode_responses=True)


async def project_stock_changes():
    consumer = AIOKafkaConsumer(
        "mysql.inventory.stock",
        bootstrap_servers="kafka:9092",
        group_id="inventory-projector",
    )
    await consumer.start()
    async for msg in consumer:
        event = json.loads(msg.value)
        if event["op"] == "u":
            r.hset(f"stock:{event['after']['sku']}", mapping={
                "qty": event["after"]["qty"],
                "updated_at": event["ts"],
            })

# ❌ 反例:用 2PC/XA 解决跨服务一致性
# 伪代码:Seata AT 模式
@GlobalTransactional  # ← 反模式!让微服务退化成分布式单体
def create_order_with_payment(order, payment):
    order_service.create(order)
    payment_service.charge(payment)  # 这行失败 → 回滚上一行
    inventory_service.deduct(...)    # 这行失败 → 回滚上面两行

# user_service/db.py
from sqlalchemy import create_engine
user_engine = create_engine("mysql+pymysql://user:pass@mysql/users_db")

# order_service/db.py
order_engine = create_engine("mysql+pymysql://user:pass@mysql/orders_db")

# inventory_service/db.py
inv_engine = create_engine("mysql+pymysql://user:pass@mysql/inventory_db")

# user_service/cache.py  # 每个服务用自己的 Redis 命名空间
import redis
user_redis = redis.Redis(host="redis", port=6379, db=0)

# inventory_service/cache.py
inv_redis = redis.Redis(host="redis", port=6379, db=1)

# formula:服务数量 ≈ 团队数量
n_services = n_teams
# 50 人 → 8-10 个服务
# 200 人 → 25-30 个服务
# 500 人 → 60-80 个服务

# Brooks 定律的微服务版本
communication_cost = O(n_teams^2)
# 8 个团队 → 64 条沟通链路
# 20 个团队 → 400 条沟通链路
# 30 个团队 → 900 条沟通链路(运维灾难)

# ❌ 反例:30 个服务,2 个团队
services_owned_by_team_A = [
    "user-svc", "order-svc", "payment-svc",
    "inventory-svc", "logistics-svc", "notification-svc",
    "coupon-svc", "report-svc", "search-svc",
    "recommend-svc", "auth-svc", "profile-svc",
    "address-svc", "cart-svc", "checkout-svc",
    # ... 15 more
]
# 每个工程师要懂 15 个服务

# CODEOWNERS 范例
# .github/CODEOWNERS

/order-service/      @team-orders
/payment-service/    @team-payments
/inventory-service/  @team-inventory
/logistics-service/  @team-logistics
/user-service/       @team-users

/shared-libs/        @team-platform
/api-gateway/        @team-platform
/observability/      @team-sre

# 评估函数:服务数量 ≈ 团队数量 / 2
def max_services(n_teams: int) -> int:
    """每个团队 1-2 个服务,不允许 1 人拥有 3 个服务"""
    return n_teams * 2

# 例:50 人 8 个团队 → 最多 16 个服务
# 例:30 人 5 个团队 → 最多 10 个服务

# 步骤 1:确认共享表
shared_tables = audit_shared_tables()  # 扫所有服务代码,找跨服务 SQL

# 步骤 2:定义新表归属
table_ownership = {
    "users": "user-service",
    "orders": "order-service",
    "stock": "inventory-service",
}

# 步骤 3:双写过渡期
def write_user(data):
    # 写主服务
    primary_db.execute("INSERT INTO users ...", data)
    # 双写到旧共享库
    legacy_db.execute("INSERT INTO legacy_users ...", data)
    # 发布事件,让数据消费者迁移
    kafka.publish("user.created", data)

# 步骤 4:校验一致
def assert_consistency():
    primary = primary_db.fetch("SELECT * FROM users WHERE id=1")
    legacy = legacy_db.fetch("SELECT * FROM legacy_users WHERE id=1")
    assert primary.email == legacy.email

# ❌ 同步长链
def create_order_sync(order):
    inv = inventory_client.reserve(order.sku, order.qty)   # RTT 1
    pay = payment_client.charge(order.amount)              # RTT 2
    logi = logistics_client.create_shipment(order.id)      # RTT 3
    noti = notification_client.send_sms(order.user_id)     # RTT 4
    return {"inv": inv, "pay": pay, "logi": logi, "noti": noti}
    # 总耗时:80ms × 4 + 网络 10ms × 4 = 360ms

# ✅ 异步事件
async def create_order_async(order):
    order = await order_repo.create(order)
    await event_bus.publish("order.created", order.to_event())
    return order  # 50ms 返回,后续异步处理

# 反例:2PC
@transactional  # XA 事务
def transfer_money(from_account, to_account, amount):
    account_dao.debit(from_account, amount)
    account_dao.credit(to_account, amount)

# 正例:Saga + 幂等 + 对账
class TransferSaga:
    """转账 Saga:扣款 → 加款 → 通知"""
    def step_1_debit(self):
        # 幂等:用 transfer_id 防重
        if self.is_done("debit"):
            return
        payment_client.hold(self.from_account, self.amount, idempotency_key=self.transfer_id)
        self.mark_done("debit")

    def step_2_credit(self):
        if self.is_done("credit"):
            return
        payment_client.credit(self.to_account, self.amount, idempotency_key=self.transfer_id)
        self.mark_done("credit")

    def compensate(self):
        # 失败补偿:把扣的钱退回去
        if self.is_done("debit") and not self.is_done("credit"):
            payment_client.release(self.from_account, self.amount, self.transfer_id)

# 对账:每日凌晨跑一次,补齐数据
def reconcile():
    pending = saga_repo.find_pending(timeout_minutes=30)
    for saga in pending:
        saga.compensate()

# 评估每个服务的"复杂度指数"
@dataclass
class ServiceMetrics:
    name: str
    lines_of_code: int
    api_count: int
    table_count: int
    team_size: int
    deploy_freq_per_day: float

def health_score(m: ServiceMetrics) -> float:
    """分越低越健康,> 70 要重新拆分"""
    score = 0
    score += min(m.lines_of_code / 1000, 50)        # 权重 50
    score += min(m.api_count, 20) * 1                # 权重 20
    score += min(m.table_count, 15) * 1              # 权重 15
    score += max(0, m.team_size - 8) * 3             # 团队>8 加分
    if m.deploy_freq_per_day < 1:
        score += 10                                  # 部署不频繁扣分
    return score

# api-gateway/kong.yml
services:
  - name: user-svc
    url: http://user-service:8000
    routes:
      - name: user-route
        paths: [/api/users]
  - name: order-svc
    url: http://order-service:8000
    routes:
      - name: order-route
        paths: [/api/orders]
  - name: payment-svc
    url: http://payment-service:8000
    routes:
      - name: payment-route
        paths: [/api/payments]

plugins:
  - name: jwt  # 统一鉴权
  - name: rate-limiting
    config:
      minute: 100
  - name: cors
  - name: prometheus  # 统一指标