028-微服务架构
·
028-微服务架构
难度：🔴 | 预计时间：120分钟 | 前置：027-异步编程
学习目标
- 理解微服务架构的核心概念和设计原则
- 掌握使用FastAPI构建微服务的最佳实践
- 学会实现服务注册与发现机制
- 掌握微服务间通信和数据一致性处理
- 了解微服务的监控、日志和错误处理策略
微服务架构概述
什么是微服务架构
微服务架构是一种将单一应用程序开发为一组小型服务的方法，每个服务运行在自己的进程中，并使用轻量级机制（通常是HTTP API）进行通信。这些服务围绕业务功能构建，并且可以由全自动部署机制独立部署。
微服务架构的优势
| 优势 | 说明 | 适用场景 |
|---|---|---|
| 技术多样性 | 每个服务可以使用不同的技术栈 | 团队技术栈多样化 |
| 独立部署 | 服务可以独立开发、测试和部署 | 快速迭代需求 |
| 故障隔离 | 单个服务的故障不会影响整个系统 | 高可用性要求 |
| 团队自治 | 小团队可以独立负责特定服务 | 大型开发团队 |
| 可扩展性 | 可以针对特定服务进行水平扩展 | 不同服务负载差异大 |
微服务架构的挑战
⚠️ 注意：微服务架构虽然带来了很多优势，但也引入了分布式系统的复杂性。
- 网络延迟：服务间通信增加了网络开销
- 数据一致性：分布式事务处理复杂
- 服务治理：需要完善的监控和管理机制
- 运维复杂度：部署和维护成本增加
微服务架构设计
系统架构图
服务拆分原则
- 业务边界清晰：按照业务领域进行服务拆分
- 数据独立性：每个服务拥有独立的数据存储
- 接口稳定性：服务间接口应该保持向后兼容
- 团队组织结构：服务边界与团队边界保持一致
核心组件实现
1. 服务注册与发现
服务注册与发现是微服务架构的核心组件，用于管理服务实例的生命周期。
# app/core/microservice.py
from pydantic import BaseModel
from typing import Dict, Any, Optional
from enum import Enum
from datetime import datetime
class ServiceStatus(str, Enum):
"""服务状态枚举"""
STARTING = "starting" # 启动中
HEALTHY = "healthy" # 健康
UNHEALTHY = "unhealthy" # 不健康
STOPPING = "stopping" # 停止中
STOPPED = "stopped" # 已停止
class ServiceInfo(BaseModel):
"""服务信息模型"""
name: str # 服务名称
version: str # 服务版本
host: str # 服务主机
port: int # 服务端口
health_endpoint: str = "/health" # 健康检查端点
metadata: Dict[str, Any] = {} # 服务元数据
status: ServiceStatus = ServiceStatus.STARTING
registered_at: Optional[datetime] = None
last_heartbeat: Optional[datetime] = None
@property
def service_url(self) -> str:
"""获取服务URL"""
return f"http://{self.host}:{self.port}"
@property
def health_url(self) -> str:
"""获取健康检查URL"""
return f"{self.service_url}{self.health_endpoint}"
# app/core/service_registry.py
import asyncio
import aiohttp
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from app.core.microservice import ServiceInfo, ServiceStatus
from app.core.logger import app_logger
class ServiceRegistry:
"""服务注册中心"""
def __init__(self):
self._services: Dict[str, ServiceInfo] = {}
self._health_check_interval = 30 # 健康检查间隔（秒）
self._health_check_task: Optional[asyncio.Task] = None
self._session: Optional[aiohttp.ClientSession] = None
async def initialize(self):
"""初始化服务注册中心"""
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=5)
)
# 启动健康检查任务
self._health_check_task = asyncio.create_task(
self._health_check_loop()
)
app_logger.info("Service registry initialized")
async def shutdown(self):
"""关闭服务注册中心"""
if self._health_check_task:
self._health_check_task.cancel()
try:
await self._health_check_task
except asyncio.CancelledError:
pass
if self._session:
await self._session.close()
app_logger.info("Service registry shutdown")
async def register_service(self, service_info: ServiceInfo) -> bool:
"""注册服务"""
try:
service_info.registered_at = datetime.utcnow()
service_info.last_heartbeat = datetime.utcnow()
service_info.status = ServiceStatus.HEALTHY
self._services[service_info.name] = service_info
app_logger.info(
f"Service registered: {service_info.name} at {service_info.service_url}",
extra={
"event_type": "service_registered",
"service_name": service_info.name,
"service_url": service_info.service_url
}
)
return True
except Exception as e:
app_logger.error(
f"Failed to register service {service_info.name}: {str(e)}",
extra={
"event_type": "service_registration_failed",
"service_name": service_info.name,
"error": str(e)
}
)
return False
async def unregister_service(self, service_name: str) -> bool:
"""注销服务"""
try:
if service_name in self._services:
del self._services[service_name]
app_logger.info(
f"Service unregistered: {service_name}",
extra={
"event_type": "service_unregistered",
"service_name": service_name
}
)
return True
return False
except Exception as e:
app_logger.error(
f"Failed to unregister service {service_name}: {str(e)}",
extra={
"event_type": "service_unregistration_failed",
"service_name": service_name,
"error": str(e)
}
)
return False
async def get_service(self, service_name: str) -> Optional[ServiceInfo]:
"""获取服务信息"""
service = self._services.get(service_name)
if service and service.status == ServiceStatus.HEALTHY:
return service
return None
async def get_all_services(self) -> List[ServiceInfo]:
"""获取所有健康的服务"""
return [
service for service in self._services.values()
if service.status == ServiceStatus.HEALTHY
]
async def _health_check_loop(self):
"""健康检查循环"""
while True:
try:
await asyncio.sleep(self._health_check_interval)
await self._perform_health_checks()
except asyncio.CancelledError:
break
except Exception as e:
app_logger.error(
f"Health check loop error: {str(e)}",
extra={
"event_type": "health_check_error",
"error": str(e)
}
)
async def _perform_health_checks(self):
"""执行健康检查"""
for service_name, service_info in list(self._services.items()):
try:
await self._check_service_health(service_info)
except Exception as e:
app_logger.error(
f"Health check failed for {service_name}: {str(e)}",
extra={
"event_type": "health_check_failed",
"service_name": service_name,
"error": str(e)
}
)
async def _check_service_health(self, service_info: ServiceInfo):
"""检查单个服务健康状态"""
try:
async with self._session.get(service_info.health_url) as response:
if response.status == 200:
service_info.status = ServiceStatus.HEALTHY
service_info.last_heartbeat = datetime.utcnow()
else:
service_info.status = ServiceStatus.UNHEALTHY
except Exception:
service_info.status = ServiceStatus.UNHEALTHY
# 如果服务长时间不健康，则移除
if (service_info.last_heartbeat and
datetime.utcnow() - service_info.last_heartbeat > timedelta(minutes=5)):
app_logger.warning(
f"Removing unhealthy service: {service_info.name}",
extra={
"event_type": "service_removed",
"service_name": service_info.name,
"reason": "unhealthy_timeout"
}
)
del self._services[service_info.name]
# 全局服务注册中心实例
service_registry = ServiceRegistry()
2. 服务间通信客户端
# app/clients/base_client.py
import aiohttp
import asyncio
from typing import Dict, Any, Optional
from app.core.service_registry import service_registry
from app.core.logger import app_logger
class BaseServiceClient:
"""基础服务客户端"""
def __init__(self, service_name: str):
self.service_name = service_name
self._session: Optional[aiohttp.ClientSession] = None
self._base_url: Optional[str] = None
async def initialize(self):
"""初始化客户端"""
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=30)
)
await self._discover_service()
async def shutdown(self):
"""关闭客户端"""
if self._session:
await self._session.close()
async def _discover_service(self):
"""服务发现"""
service_info = await service_registry.get_service(self.service_name)
if service_info:
self._base_url = service_info.service_url
else:
raise RuntimeError(f"Service {self.service_name} not found")
async def _request(
self,
method: str,
endpoint: str,
**kwargs
) -> Dict[str, Any]:
"""发送HTTP请求"""
if not self._session or not self._base_url:
await self._discover_service()
url = f"{self._base_url}{endpoint}"
try:
async with self._session.request(method, url, **kwargs) as response:
if response.status >= 400:
error_text = await response.text()
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=error_text
)
return await response.json()
except aiohttp.ClientError as e:
app_logger.error(
f"Request to {self.service_name} failed: {str(e)}",
extra={
"event_type": "service_request_failed",
"service_name": self.service_name,
"endpoint": endpoint,
"error": str(e)
}
)
# 重新发现服务
await self._discover_service()
raise
async def get(self, endpoint: str, **kwargs) -> Dict[str, Any]:
"""GET请求"""
return await self._request("GET", endpoint, **kwargs)
async def post(self, endpoint: str, **kwargs) -> Dict[str, Any]:
"""POST请求"""
return await self._request("POST", endpoint, **kwargs)
async def put(self, endpoint: str, **kwargs) -> Dict[str, Any]:
"""PUT请求"""
return await self._request("PUT", endpoint, **kwargs)
async def delete(self, endpoint: str, **kwargs) -> Dict[str, Any]:
"""DELETE请求"""
return await self._request("DELETE", endpoint, **kwargs)
# app/clients/user_client.py
from typing import Dict, Any, Optional
from app.clients.base_client import BaseServiceClient
class UserClient(BaseServiceClient):
"""用户服务客户端"""
def __init__(self):
super().__init__("user-service")
async def get_user(self, user_id: int) -> Optional[Dict[str, Any]]:
"""获取用户信息"""
try:
return await self.get(f"/users/{user_id}")
except Exception:
return None
async def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""创建用户"""
return await self.post("/users", json=user_data)
async def update_user(
self,
user_id: int,
user_data: Dict[str, Any]
) -> Dict[str, Any]:
"""更新用户信息"""
return await self.put(f"/users/{user_id}", json=user_data)
async def verify_user_credentials(
self,
username: str,
password: str
) -> Optional[Dict[str, Any]]:
"""验证用户凭据"""
try:
return await self.post(
"/auth/verify",
json={"username": username, "password": password}
)
except Exception:
return None
# 全局用户客户端实例
user_client = UserClient()
3. 文章服务实现
# services/post_service/main.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
# 导入共享模块
from app.core.microservice import ServiceInfo, ServiceStatus
from app.core.service_registry import service_registry
from app.core.logger import app_logger
from app.core.config import settings
from app.models.post import Post, PostStatus
from app.services.post_service import PostService
from app.core.database import get_db
from app.clients.user_client import user_client
# 创建FastAPI应用
app = FastAPI(
title="Post Service",
description="文章管理微服务",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# 添加CORS中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 服务信息
SERVICE_INFO = ServiceInfo(
name="post-service",
version="1.0.0",
host=settings.POST_SERVICE_HOST,
port=settings.POST_SERVICE_PORT,
health_endpoint="/health",
metadata={
"description": "文章管理服务",
"capabilities": ["post_management", "content_management", "publishing"]
}
)
# Pydantic模型
class PostCreate(BaseModel):
"""创建文章请求模型"""
title: str
content: str
summary: Optional[str] = None
tags: Optional[List[str]] = None
status: PostStatus = PostStatus.DRAFT
class PostUpdate(BaseModel):
"""更新文章请求模型"""
title: Optional[str] = None
content: Optional[str] = None
summary: Optional[str] = None
tags: Optional[List[str]] = None
status: Optional[PostStatus] = None
class PostResponse(BaseModel):
"""文章响应模型"""
id: int
title: str
content: str
summary: Optional[str]
slug: str
status: PostStatus
author_id: int
author: Optional[Dict[str, Any]] = None # 来自用户服务的数据
tags: Optional[List[str]]
view_count: int
like_count: int
comment_count: int
created_at: datetime
updated_at: datetime
published_at: Optional[datetime]
class Config:
from_attributes = True
class HealthResponse(BaseModel):
"""健康检查响应模型"""
status: str
timestamp: datetime
service: str
version: str
# 健康检查端点
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""健康检查"""
return HealthResponse(
status="healthy",
timestamp=datetime.utcnow(),
service=SERVICE_INFO.name,
version=SERVICE_INFO.version
)
# 文章管理端点
@app.post("/posts", response_model=PostResponse)
async def create_post(
post_data: PostCreate,
author_id: int, # 通常从JWT token中获取
db: AsyncSession = Depends(get_db)
):
"""创建文章"""
try:
# 验证作者是否存在（调用用户服务）
author = await user_client.get_user(author_id)
if not author:
raise HTTPException(status_code=400, detail="Invalid author")
post_service = PostService(db)
# 创建文章
post = await post_service.create_post(
title=post_data.title,
content=post_data.content,
summary=post_data.summary,
author_id=author_id,
tags=post_data.tags,
status=post_data.status
)
# 构建响应，包含作者信息
response_data = PostResponse.from_orm(post)
response_data.author = author
app_logger.info(
f"Post created: {post.title}",
extra={
"event_type": "post_created",
"post_id": post.id,
"author_id": author_id,
"title": post.title
}
)
return response_data
except HTTPException:
raise
except Exception as e:
app_logger.error(
f"Failed to create post: {str(e)}",
extra={
"event_type": "post_creation_failed",
"author_id": author_id,
"error": str(e)
}
)
raise HTTPException(status_code=500, detail="Internal server error")
# 应用生命周期事件
@app.on_event("startup")
async def startup_event():
"""应用启动事件"""
try:
# 初始化服务注册中心
await service_registry.initialize()
# 初始化用户客户端
await user_client.initialize()
# 注册服务
await service_registry.register_service(SERVICE_INFO)
app_logger.info(
f"Post service started on {SERVICE_INFO.host}:{SERVICE_INFO.port}",
extra={
"event_type": "service_started",
"service_name": SERVICE_INFO.name,
"service_version": SERVICE_INFO.version
}
)
except Exception as e:
app_logger.error(
f"Failed to start post service: {str(e)}",
extra={
"event_type": "service_startup_failed",
"error": str(e)
}
)
raise
@app.on_event("shutdown")
async def shutdown_event():
"""应用关闭事件"""
try:
# 关闭用户客户端
await user_client.shutdown()
# 注销服务
await service_registry.unregister_service(SERVICE_INFO.name)
# 关闭服务注册中心
await service_registry.shutdown()
app_logger.info(
"Post service shutdown",
extra={
"event_type": "service_shutdown",
"service_name": SERVICE_INFO.name
}
)
except Exception as e:
app_logger.error(
f"Error during post service shutdown: {str(e)}",
extra={
"event_type": "service_shutdown_error",
"error": str(e)
}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host=SERVICE_INFO.host,
port=SERVICE_INFO.port,
reload=True
)
数据一致性处理
分布式事务模式
在微服务架构中，数据一致性是一个重要挑战。我们可以使用以下模式来处理：
1. Saga模式
# app/core/saga.py
from typing import List, Callable, Any, Dict
from enum import Enum
import asyncio
from app.core.logger import app_logger
class SagaStepStatus(str, Enum):
"""Saga步骤状态"""
PENDING = "pending"
COMPLETED = "completed"
FAILED = "failed"
COMPENSATED = "compensated"
class SagaStep:
"""Saga步骤"""
def __init__(
self,
name: str,
action: Callable,
compensation: Callable,
**kwargs
):
self.name = name
self.action = action
self.compensation = compensation
self.kwargs = kwargs
self.status = SagaStepStatus.PENDING
self.result = None
self.error = None
class SagaOrchestrator:
"""Saga编排器"""
def __init__(self, saga_id: str):
self.saga_id = saga_id
self.steps: List[SagaStep] = []
self.completed_steps: List[SagaStep] = []
def add_step(
self,
name: str,
action: Callable,
compensation: Callable,
**kwargs
):
"""添加Saga步骤"""
step = SagaStep(name, action, compensation, **kwargs)
self.steps.append(step)
return self
async def execute(self) -> bool:
"""执行Saga"""
app_logger.info(
f"Starting saga {self.saga_id}",
extra={
"event_type": "saga_started",
"saga_id": self.saga_id,
"steps_count": len(self.steps)
}
)
try:
# 顺序执行所有步骤
for step in self.steps:
await self._execute_step(step)
self.completed_steps.append(step)
app_logger.info(
f"Saga {self.saga_id} completed successfully",
extra={
"event_type": "saga_completed",
"saga_id": self.saga_id
}
)
return True
except Exception as e:
app_logger.error(
f"Saga {self.saga_id} failed: {str(e)}",
extra={
"event_type": "saga_failed",
"saga_id": self.saga_id,
"error": str(e)
}
)
# 执行补偿操作
await self._compensate()
return False
async def _execute_step(self, step: SagaStep):
"""执行单个步骤"""
try:
app_logger.info(
f"Executing saga step: {step.name}",
extra={
"event_type": "saga_step_started",
"saga_id": self.saga_id,
"step_name": step.name
}
)
step.result = await step.action(**step.kwargs)
step.status = SagaStepStatus.COMPLETED
app_logger.info(
f"Saga step completed: {step.name}",
extra={
"event_type": "saga_step_completed",
"saga_id": self.saga_id,
"step_name": step.name
}
)
except Exception as e:
step.status = SagaStepStatus.FAILED
step.error = str(e)
raise
async def _compensate(self):
"""执行补偿操作"""
app_logger.info(
f"Starting compensation for saga {self.saga_id}",
extra={
"event_type": "saga_compensation_started",
"saga_id": self.saga_id
}
)
# 逆序执行补偿操作
for step in reversed(self.completed_steps):
try:
await step.compensation(step.result, **step.kwargs)
step.status = SagaStepStatus.COMPENSATED
app_logger.info(
f"Compensated saga step: {step.name}",
extra={
"event_type": "saga_step_compensated",
"saga_id": self.saga_id,
"step_name": step.name
}
)
except Exception as e:
app_logger.error(
f"Failed to compensate step {step.name}: {str(e)}",
extra={
"event_type": "saga_compensation_failed",
"saga_id": self.saga_id,
"step_name": step.name,
"error": str(e)
}
)
2. 事件驱动架构
# app/core/events.py
from pydantic import BaseModel
from typing import Any, Dict, List, Callable
from datetime import datetime
import asyncio
from app.core.logger import app_logger
class DomainEvent(BaseModel):
"""领域事件基类"""
event_id: str
event_type: str
aggregate_id: str
aggregate_type: str
event_data: Dict[str, Any]
occurred_at: datetime
version: int = 1
class EventHandler:
"""事件处理器基类"""
def __init__(self, event_type: str):
self.event_type = event_type
async def handle(self, event: DomainEvent):
"""处理事件"""
raise NotImplementedError
class EventBus:
"""事件总线"""
def __init__(self):
self._handlers: Dict[str, List[EventHandler]] = {}
def subscribe(self, event_type: str, handler: EventHandler):
"""订阅事件"""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
async def publish(self, event: DomainEvent):
"""发布事件"""
handlers = self._handlers.get(event.event_type, [])
app_logger.info(
f"Publishing event: {event.event_type}",
extra={
"event_type": "domain_event_published",
"domain_event_type": event.event_type,
"aggregate_id": event.aggregate_id,
"handlers_count": len(handlers)
}
)
# 并行处理所有事件处理器
tasks = []
for handler in handlers:
task = asyncio.create_task(self._handle_event(handler, event))
tasks.append(task)
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
async def _handle_event(self, handler: EventHandler, event: DomainEvent):
"""处理单个事件"""
try:
await handler.handle(event)
app_logger.info(
f"Event handled successfully: {event.event_type} by {handler.__class__.__name__}",
extra={
"event_type": "domain_event_handled",
"domain_event_type": event.event_type,
"handler": handler.__class__.__name__,
"aggregate_id": event.aggregate_id
}
)
except Exception as e:
app_logger.error(
f"Failed to handle event {event.event_type}: {str(e)}",
extra={
"event_type": "domain_event_handling_failed",
"domain_event_type": event.event_type,
"handler": handler.__class__.__name__,
"aggregate_id": event.aggregate_id,
"error": str(e)
}
)
# 全局事件总线实例
event_bus = EventBus()
监控和日志
结构化日志
# app/core/logger.py
import logging
import json
from datetime import datetime
from typing import Dict, Any, Optional
class StructuredFormatter(logging.Formatter):
"""结构化日志格式化器"""
def format(self, record: logging.LogRecord) -> str:
# 基础日志信息
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno
}
# 添加额外的上下文信息
if hasattr(record, 'extra') and record.extra:
log_data.update(record.extra)
# 添加异常信息
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
return json.dumps(log_data, ensure_ascii=False)
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""设置结构化日志记录器"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper()))
# 避免重复添加处理器
if not logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger.addHandler(handler)
return logger
# 应用日志记录器
app_logger = setup_logger("microservice")
性能监控
# app/core/metrics.py
import time
from typing import Dict, Any, Optional
from functools import wraps
from app.core.logger import app_logger
class MetricsCollector:
"""指标收集器"""
def __init__(self):
self._counters: Dict[str, int] = {}
self._gauges: Dict[str, float] = {}
self._histograms: Dict[str, list] = {}
def increment_counter(self, name: str, value: int = 1, tags: Optional[Dict[str, str]] = None):
"""增加计数器"""
key = self._build_key(name, tags)
self._counters[key] = self._counters.get(key, 0) + value
def set_gauge(self, name: str, value: float, tags: Optional[Dict[str, str]] = None):
"""设置仪表盘值"""
key = self._build_key(name, tags)
self._gauges[key] = value
def record_histogram(self, name: str, value: float, tags: Optional[Dict[str, str]] = None):
"""记录直方图值"""
key = self._build_key(name, tags)
if key not in self._histograms:
self._histograms[key] = []
self._histograms[key].append(value)
def _build_key(self, name: str, tags: Optional[Dict[str, str]] = None) -> str:
"""构建指标键"""
if not tags:
return name
tag_str = ",".join([f"{k}={v}" for k, v in sorted(tags.items())])
return f"{name}[{tag_str}]"
def get_metrics(self) -> Dict[str, Any]:
"""获取所有指标"""
return {
"counters": self._counters.copy(),
"gauges": self._gauges.copy(),
"histograms": {
k: {
"count": len(v),
"sum": sum(v),
"avg": sum(v) / len(v) if v else 0,
"min": min(v) if v else 0,
"max": max(v) if v else 0
}
for k, v in self._histograms.items()
}
}
# 全局指标收集器
metrics = MetricsCollector()
def monitor_performance(metric_name: str, tags: Optional[Dict[str, str]] = None):
"""性能监控装饰器"""
def decorator(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = await func(*args, **kwargs)
# 记录成功指标
metrics.increment_counter(
f"{metric_name}.success",
tags=tags
)
return result
except Exception as e:
# 记录失败指标
metrics.increment_counter(
f"{metric_name}.error",
tags={**(tags or {}), "error_type": type(e).__name__}
)
raise
finally:
# 记录执行时间
duration = time.time() - start_time
metrics.record_histogram(
f"{metric_name}.duration",
duration,
tags=tags
)
app_logger.info(
f"Function {func.__name__} executed in {duration:.3f}s",
extra={
"event_type": "function_performance",
"function_name": func.__name__,
"duration": duration,
"tags": tags
}
)
@wraps(func)
def sync_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
metrics.increment_counter(
f"{metric_name}.success",
tags=tags
)
return result
except Exception as e:
metrics.increment_counter(
f"{metric_name}.error",
tags={**(tags or {}), "error_type": type(e).__name__}
)
raise
finally:
duration = time.time() - start_time
metrics.record_histogram(
f"{metric_name}.duration",
duration,
tags=tags
)
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
return decorator
部署和运维
Docker容器化
# Dockerfile
FROM python:3.11-slim
# 设置工作目录
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建非root用户
RUN useradd --create-home --shell /bin/bash app \
&& chown -R app:app /app
USER app
# 暴露端口
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
Docker Compose配置
# docker-compose.yml
version: '3.8'
services:
# 用户服务
user-service:
build:
context: ./services/user_service
dockerfile: Dockerfile
ports:
- "8001:8000"
environment:
- DATABASE_URL=postgresql+asyncpg://user:password@postgres:5432/userdb
- REDIS_URL=redis://redis:6379/0
depends_on:
- postgres
- redis
networks:
- microservices
restart: unless-stopped
# 文章服务
post-service:
build:
context: ./services/post_service
dockerfile: Dockerfile
ports:
- "8002:8000"
environment:
- DATABASE_URL=postgresql+asyncpg://user:password@postgres:5432/postdb
- REDIS_URL=redis://redis:6379/1
- USER_SERVICE_URL=http://user-service:8000
depends_on:
- postgres
- redis
- user-service
networks:
- microservices
restart: unless-stopped
# API网关
api-gateway:
build:
context: ./gateway
dockerfile: Dockerfile
ports:
- "8000:8000"
environment:
- USER_SERVICE_URL=http://user-service:8000
- POST_SERVICE_URL=http://post-service:8000
depends_on:
- user-service
- post-service
networks:
- microservices
restart: unless-stopped
# 数据库
postgres:
image: postgres:15
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=microservices
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
networks:
- microservices
restart: unless-stopped
# 缓存
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
networks:
- microservices
restart: unless-stopped
# 消息队列
rabbitmq:
image: rabbitmq:3-management
environment:
- RABBITMQ_DEFAULT_USER=admin
- RABBITMQ_DEFAULT_PASS=password
ports:
- "15672:15672" # 管理界面
volumes:
- rabbitmq_data:/var/lib/rabbitmq
networks:
- microservices
restart: unless-stopped
volumes:
postgres_data:
redis_data:
rabbitmq_data:
networks:
microservices:
driver: bridge
实践练习
练习1：实现用户服务
- 创建用户服务的完整实现（验收标准：包含用户注册、登录、信息管理功能）
- 实现JWT认证机制（验收标准：能够生成和验证JWT token）
- 添加用户权限管理（验收标准：支持角色和权限控制）
练习2：实现API网关
- 创建API网关服务（验收标准：能够路由请求到不同的微服务）
- 实现请求认证和授权（验收标准：统一的认证入口）
- 添加请求限流和熔断机制（验收标准：能够防止服务过载）
练习3：实现分布式事务
- 使用Saga模式实现订单创建流程（验收标准：包含库存扣减、支付、订单创建的完整流程）
- 实现补偿机制（验收标准：任何步骤失败都能正确回滚）
- 添加事务状态监控（验收标准：能够追踪事务执行状态）
练习4：监控和运维
- 集成Prometheus指标收集（验收标准：能够收集服务性能指标）
- 实现分布式链路追踪（验收标准：能够追踪跨服务的请求链路）
- 配置日志聚合系统（验收标准：集中化的日志管理）
常见问题
Q: 微服务拆分的粒度如何把握？
A: 遵循单一职责原则，一个服务应该只负责一个业务领域。过细的拆分会增加系统复杂度，过粗的拆分失去了微服务的优势。建议从单体应用开始，逐步拆分。
Q: 如何处理微服务间的数据一致性？
A: 优先使用最终一致性，通过事件驱动架构和Saga模式来处理分布式事务。避免使用分布式锁和两阶段提交等强一致性方案。
Q: 微服务的性能开销如何优化？
A: 1) 使用连接池减少连接开销；2) 实现智能路由和负载均衡；3) 添加缓存层；4) 使用异步通信；5) 合理设计API减少网络调用次数。
Q: 如何保证微服务的高可用性？
A: 1) 实现健康检查和自动故障转移；2) 使用熔断器模式防止级联故障；3) 部署多个实例实现冗余；4) 实现优雅降级机制；5) 定期进行故障演练。
总结
微服务架构是一种强大的架构模式，但也带来了额外的复杂性。在实施微服务架构时，需要考虑以下关键点：
- 服务拆分：基于业务边界进行合理拆分
- 服务治理：实现完善的服务注册、发现和监控机制
- 数据一致性：使用事件驱动和Saga模式处理分布式事务
- 可观测性：建立完善的日志、指标和链路追踪体系
- 运维自动化：使用容器化和CI/CD实现自动化部署
💡 提示：微服务架构不是银弹，需要根据团队规模、业务复杂度和技术能力来决定是否采用。
下一步
参考与引用
更新记录
- 更新时间: 2024-01-15 | 更新内容: 创建微服务架构教程，包含完整的理论和实践内容 | 更新人: Assistant
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
更多推荐
28
0- 0
已为社区贡献3条内容
所有评论(0)