Python 微服务架构：从单体拆分到服务治理

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="User Service", version="1.0.0")

app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"])

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/api/v1/users/{user_id}")
def get_user(user_id: int):
    user = user_repo.get(user_id)
    if user is None:
        raise HTTPException(404, "用户不存在")
    return user

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://...")
Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_user(db: AsyncSession, user_id: int):
    return await db.get(User, user_id)

import httpx

class OrderServiceClient:
    def __init__(self, base_url: str):
        self.client = httpx.AsyncClient(base_url=base_url, timeout=5)

    async def create_order(self, payload: dict) -> dict:
        r = await self.client.post("/api/v1/orders", json=payload)
        r.raise_for_status()      # 失败快速暴露
        return r.json()

service UserService {
  rpc GetUser (GetUserRequest) returns (UserReply);
}

import grpc
import user_pb2, user_pb2_grpc

channel = grpc.aio.insecure_channel("user-service:50051")
stub = user_pb2_grpc.UserServiceStub(channel)

async def get_user(user_id: int):
    resp = await stub.GetUser(user_pb2.GetUserRequest(id=user_id))
    return resp

import pika

def publish(channel, exchange, routing_key, payload):
    channel.basic_publish(
        exchange=exchange, routing_key=routing_key,
        body=json.dumps(payload).encode(),
        properties=pika.BasicProperties(
            delivery_mode=2,       # 持久化
            content_type="application/json",
        ),
    )

def consume(channel, queue, callback):
    channel.basic_qos(prefetch_count=1)   # 一次一个，公平分发
    channel.basic_consume(queue, callback, auto_ack=False)
    channel.start_consuming()

from kafka import KafkaProducer, KafkaConsumer
import json

producer = KafkaProducer(
    bootstrap_servers="kafka:9092",
    value_serializer=lambda v: json.dumps(v).encode(),
)

producer.send("order-events", {"order_id": 1, "status": "created"})

consumer = KafkaConsumer(
    "order-events",
    bootstrap_servers="kafka:9092",
    group_id="notification-service",
    auto_offset_reset="earliest",
)
for msg in consumer:
    event = json.loads(msg.value)
    handle_event(event)

# 用 traefik / kong 或 FastAPI 自建
from fastapi import FastAPI
import httpx

app = FastAPI()
routes = {
    "/users": "http://user-service:8000",
    "/orders": "http://order-service:8000",
}

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(path: str):
    target = routes.get("/" + path.split("/")[0])
    async with httpx.AsyncClient() as client:
        resp = await client.request(...)
    return resp.json()

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "user-service"
    database_url: str
    kafka_brokers: str
    log_level: str = "INFO"

    model_config = {"env_file": ".env", "env_prefix": ""}

settings = Settings()

# Kubernetes Secret 引用
env:
  - name: DATABASE_PASSWORD
    valueFrom:
      secretKeyRef:
        name: db-cred
        key: password

import structlog, logging

structlog.configure(processors=[
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.JSONRenderer(),
])
log = structlog.get_logger()
log.info("order.created", order_id=1, service="order-service")

from prometheus_client import Counter, Histogram, start_http_server

REQUESTS = Counter("http_requests_total", "请求数", ["service", "status"])
LATENCY = Histogram("http_request_duration_seconds", "耗时", ["service"])

def middleware(handler):
    def wrapper(request):
        with LATENCY.labels(service="user").time():
            resp = handler(request)
        REQUESTS.labels(service="user", status=resp.status).inc()
        return resp
    return wrapper

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="otel-collector:4317")))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("handle_request"):
    # 跨服务传递 traceparent header 串联
    ...

FROM python:3.12-slim AS base
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

FROM base
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# K8s liveness/readiness 探针
@app.get("/health/live")
def live(): return {"status": "ok"}

@app.get("/health/ready")
def ready():
    db_ok = check_db()
    return {"status": "ok" if db_ok else "degraded"}

# K8s deployment
livenessProbe:
  httpGet: { path: /health/live, port: 8000 }
readinessProbe:
  httpGet: { path: /health/ready, port: 8000 }