# 🚀 大型电商平台重构工作流
## ThinkPHP 3.x → FastAPI + Ant Design Pro Vue + MySQL 8.0
---
## 📋 项目概览
### 当前系统分析
- **技术栈**: ThinkPHP 3.x + MySQL 5.7 + PHP 模板
- **规模**: 180+ 数据库表，10+ 业务模块，复杂订单系统
- **核心业务**: 电商订单管理、骑手配送、支付处理、会员系统、门店管理
- **复杂度**: 多租户架构、实时定位、多种支付方式、复杂订单状态机
### 目标架构
- **后端**: FastAPI (Python 3.9+) + SQLAlchemy + Redis
- **前端**: Ant Design Pro Vue (Vue 3 + TypeScript)
- **数据库**: MySQL 8.0 优化版本
- **架构**: 模块化单体 → 微服务演进
- **基础设施**: Docker + Nginx + CI/CD
---
## 🎯 总体迁移策略
### 核心原则
1. **零停机迁移**: 渐进式迁移，新旧系统并行运行
2. **数据完整性**: 金融级数据安全和一致性
3. **功能对等**: 所有现有功能必须完整保留
4. **渐进式实施**: 分阶段交付，降低风险
5. **性能优先**: 新系统性能必须显著优于旧系统
### 技术风险控制
- **数据库双写**: 迁移期间确保数据同步
- **API网关**: 新旧系统平滑切换
- **监控告警**: 全链路性能监控
- **回滚机制**: 快速回退到旧系统能力
---
## 📅 分阶段实施计划
## 🏗️ 第一阶段：基础设施搭建 (1-3个月)
### 1.1 环境准备与基础设施 (2周)
```yaml
目标: 搭建开发和部署环境
关键交付:
- Docker开发环境
- CI/CD流水线
- 数据库环境(新旧并行)
- 监控和日志系统
```
**具体任务清单:**
- [ ] 设置Docker开发环境 (dev/staging/prod)
- [ ] 配置GitLab CI/CD流水线
- [ ] 搭建MySQL 8.0集群并配置主从复制
- [ ] 设置Redis集群用于缓存和会话
- [ ] 配置ELK日志收集和监控系统
- [ ] 建立数据库双写同步机制
### 1.2 数据库迁移设计与实施 (3周)
```yaml
目标: 设计优化数据库结构并制定迁移策略
关键交付:
- 优化的MySQL 8.0数据库结构
- 数据迁移脚本和验证程序
- 双写同步机制
- 回滚方案
```
**核心表结构重新设计:**
```sql
-- 订单表优化设计
CREATE TABLE `orders` (
`id` BIGINT PRIMARY KEY AUTO_INCREMENT,
`order_no` VARCHAR(32) UNIQUE NOT NULL,
`merchant_id` BIGINT NOT NULL,
`store_id` BIGINT NOT NULL,
`customer_id` BIGINT,
`rider_id` BIGINT,
`order_type` TINYINT DEFAULT 1,
`status` TINYINT NOT NULL DEFAULT 0,
`payment_status` TINYINT NOT NULL DEFAULT 0,
`original_amount` DECIMAL(12,2) NOT NULL,
`discount_amount` DECIMAL(12,2) DEFAULT 0.00,
`delivery_fee` DECIMAL(8,2) DEFAULT 0.00,
`final_amount` DECIMAL(12,2) NOT NULL,
`payment_method` TINYINT,
`payment_details` JSON,
`delivery_address` JSON,
`items` JSON NOT NULL,
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
`updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`completed_at` TIMESTAMP NULL,
INDEX idx_merchant_store (merchant_id, store_id),
INDEX idx_customer (customer_id),
INDEX idx_rider (rider_id),
INDEX idx_status_created (status, created_at),
INDEX idx_order_no (order_no)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 骑手表优化设计
CREATE TABLE `riders` (
`id` BIGINT PRIMARY KEY AUTO_INCREMENT,
`merchant_id` BIGINT NOT NULL,
`store_id` BIGINT NOT NULL,
`name` VARCHAR(50) NOT NULL,
`phone` VARCHAR(20) NOT NULL,
`avatar_url` VARCHAR(255),
`id_card` VARCHAR(18) UNIQUE,
`id_card_front` VARCHAR(255),
`id_card_back` VARCHAR(255),
`health_cert` VARCHAR(255),
`health_cert_expire` DATE,
`work_status` TINYINT DEFAULT 0 COMMENT '0离线 1在线 2忙碌',
`auth_status` TINYINT DEFAULT 0 COMMENT '0未认证 1认证中 2已认证 3认证失败',
`current_location` POINT,
`location_updated_at` TIMESTAMP NULL,
`balance` DECIMAL(10,2) DEFAULT 0.00,
`frozen_balance` DECIMAL(10,2) DEFAULT 0.00,
`total_orders` INT DEFAULT 0,
`completed_orders` INT DEFAULT 0,
`rating` DECIMAL(3,2) DEFAULT 5.00,
`on_time_rate` DECIMAL(5,2) DEFAULT 100.00,
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
`updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`last_login_at` TIMESTAMP NULL,
INDEX idx_merchant_store (merchant_id, store_id),
INDEX idx_phone (phone),
INDEX idx_work_status (work_status),
INDEX idx_location (current_location),
INDEX idx_rating (rating DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
```
**数据迁移策略:**
- **双写阶段**: 新旧系统同时写入，确保数据一致性
- **验证阶段**: 定期对比新旧数据库数据差异
- **切换阶段**: 逐步切换读操作到新数据库
- **清理阶段**: 确认无误后清理旧数据库
### 1.3 FastAPI项目架构设计 (2周)
```python
# 项目结构
project_structure = """
fastapi_ecommerce/
├── app/
│ ├── core/
│ │ ├── config.py # 配置管理
│ │ ├── security.py # 安全认证
│ │ ├── database.py # 数据库连接
│ │ └── middleware.py # 中间件
│ ├── schemas/ # Pydantic数据模型
│ │ ├── order.py
│ │ ├── rider.py
│ │ ├── store.py
│ │ └── user.py
│ ├── services/ # 业务逻辑层
│ │ ├── order_service.py
│ │ ├── payment_service.py
│ │ ├── rider_service.py
│ │ └── notification_service.py
│ ├── api/ # API路由
│ │ ├── v1/
│ │ │ ├── orders.py
│ │ │ ├── riders.py
│ │ │ ├── stores.py
│ │ │ └── payments.py
│ │ └── dependencies.py # 依赖注入
│ ├── models/ # SQLAlchemy模型
│ │ ├── base.py
│ │ ├── order.py
│ │ ├── rider.py
│ │ └── store.py
│ ├── utils/ # 工具函数
│ └── tests/ # 测试文件
├── migrations/ # 数据库迁移文件
├── scripts/ # 脚本文件
└── requirements.txt
"""
# 核心配置示例
from pydantic import BaseSettings
class Settings(BaseSettings):
# 数据库配置
DATABASE_URL: str = "mysql+asyncmy://user:pass@localhost/db"
# Redis配置
REDIS_URL: str = "redis://localhost:6379"
# JWT配置
SECRET_KEY: str = "your-secret-key"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# 支付配置
WECHAT_APP_ID: str
WECHAT_MCH_ID: str
ALIPAY_APP_ID: str
class Config:
env_file = ".env"
```
### 1.4 身份认证与权限系统 (1周)
```python
# JWT + RBAC权限控制
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer
import jwt
class AuthService:
def create_access_token(self, data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=30)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_token(self, token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid token")
return user_id
except jwt.PyJWTError:
raise HTTPException(status_code=401, detail="Invalid token")
# 权限装饰器
def require_permission(permission: str):
def decorator(func):
async def wrapper(*args, **kwargs):
user = kwargs.get('current_user')
if not user.has_permission(permission):
raise HTTPException(status_code=403, detail="Permission denied")
return await func(*args, **kwargs)
return wrapper
return decorator
```
---
## 🛒 第二阶段：核心业务系统迁移 (4-6个月)
### 2.1 订单管理系统重构 (4周)
```yaml
目标: 迁移核心订单管理功能
关键交付:
- 订单CRUD操作
- 订单状态机
- 订单支付流程
- 订单退款流程
- 订单查询和统计
```
**订单状态机设计:**
```python
from enum import Enum
from typing import Dict, List
class OrderStatus(str, Enum):
PENDING = "pending" # 待支付
PAID = "paid" # 已支付
PREPARING = "preparing" # 准备中
READY = "ready" # 待配送
PICKING = "picking" # 配送中
DELIVERED = "delivered" # 已送达
COMPLETED = "completed" # 已完成
CANCELLED = "cancelled" # 已取消
REFUNDED = "refunded" # 已退款
class OrderStateMachine:
transitions = {
OrderStatus.PENDING: [OrderStatus.PAID, OrderStatus.CANCELLED],
OrderStatus.PAID: [OrderStatus.PREPARING, OrderStatus.REFUNDED],
OrderStatus.PREPARING: [OrderStatus.READY, OrderStatus.CANCELLED],
OrderStatus.READY: [OrderStatus.PICKING],
OrderStatus.PICKING: [OrderStatus.DELIVERED, OrderStatus.READY],
OrderStatus.DELIVERED: [OrderStatus.COMPLETED],
OrderStatus.COMPLETED: [OrderStatus.REFUNDED],
OrderStatus.CANCELLED: [],
OrderStatus.REFUNDED: []
}
def can_transition(self, from_status: OrderStatus, to_status: OrderStatus) -> bool:
return to_status in self.transitions.get(from_status, [])
def transition(self, current_status: OrderStatus, new_status: OrderStatus) -> bool:
if self.can_transition(current_status, new_status):
# 记录状态变更日志
# 触发相关业务逻辑
return True
return False
```
**API设计:**
```python
from fastapi import APIRouter, Depends, HTTPException
from app.schemas.order import OrderCreate, OrderResponse, OrderUpdate
from app.services.order_service import OrderService
router = APIRouter(prefix="/api/v1/orders", tags=["orders"])
@router.post("/", response_model=OrderResponse)
async def create_order(
order_data: OrderCreate,
current_user: User = Depends(get_current_user),
order_service: OrderService = Depends()
):
"""创建新订单"""
return await order_service.create_order(order_data, current_user.id)
@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(
order_id: int,
current_user: User = Depends(get_current_user)
):
"""获取订单详情"""
order = await order_service.get_order_by_id(order_id)
if not order or order.user_id != current_user.id:
raise HTTPException(status_code=404, detail="Order not found")
return order
@router.patch("/{order_id}/status")
async def update_order_status(
order_id: int,
status_data: OrderUpdate,
current_user: User = Depends(get_current_user)
):
"""更新订单状态"""
return await order_service.update_order_status(
order_id, status_data.status, current_user.id
)
```
### 2.2 支付系统现代化 (3周)
```yaml
目标: 重构支付系统，支持多种支付方式
关键交付:
- 微信支付集成
- 支付宝支付集成
- 组合支付支持
- 退款处理
- 支付回调处理
```
**支付系统架构:**
```python
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any
class PaymentProvider(ABC):
@abstractmethod
async def create_payment(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
pass
@abstractmethod
async def verify_payment(self, payment_data: Dict[str, Any]) -> bool:
pass
@abstractmethod
async def refund_payment(self, payment_id: str, amount: float) -> Dict[str, Any]:
pass
class WechatPayProvider(PaymentProvider):
def __init__(self, app_id: str, mch_id: str, api_key: str):
self.app_id = app_id
self.mch_id = mch_id
self.api_key = api_key
async def create_payment(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
# 微信支付统一下单接口
# 返回支付参数供客户端调用
pass
async def verify_payment(self, payment_data: Dict[str, Any]) -> bool:
# 验证微信支付回调
pass
class AlipayProvider(PaymentProvider):
def __init__(self, app_id: str, private_key: str, public_key: str):
self.app_id = app_id
self.private_key = private_key
self.public_key = public_key
async def create_payment(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
# 支付宝支付接口
pass
async def verify_payment(self, payment_data: Dict[str, Any]) -> bool:
# 验证支付宝回调
pass
class PaymentService:
def __init__(self):
self.providers = {
"wechat": WechatPayProvider(),
"alipay": AlipayProvider(),
"cash": CashPaymentProvider(),
"balance": BalancePaymentProvider()
}
async def process_payment(self, order_id: int, payment_method: str, amount: float):
provider = self.providers.get(payment_method)
if not provider:
raise ValueError(f"Unsupported payment method: {payment_method}")
return await provider.create_payment({
"order_id": order_id,
"amount": amount
})
```
### 2.3 门店管理系统迁移 (2周)
```python
# 门店配置管理
from app.schemas.store import StoreConfig, StoreCreate, StoreUpdate
@router.get("/stores/{store_id}/config")
async def get_store_config(store_id: int):
"""获取门店配置"""
config = await store_service.get_config(store_id)
return config
@router.put("/stores/{store_id}/config")
async def update_store_config(
store_id: int,
config: StoreConfig,
current_user: User = Depends(get_current_merchant_user)
):
"""更新门店配置"""
return await store_service.update_config(store_id, config)
@router.get("/stores/{store_id}/inventory")
async def get_store_inventory(store_id: int):
"""获取门店库存"""
return await store_service.get_inventory(store_id)
@router.post("/stores/{store_id}/delivery-areas")
async def update_delivery_areas(
store_id: int,
areas: List[DeliveryArea],
current_user: User = Depends(get_current_merchant_user)
):
"""更新门店配送范围"""
return await store_service.update_delivery_areas(store_id, areas)
```
---
## 🏍️ 第三阶段：高级功能实现 (7-9个月)
### 3.1 骑手管理与实时定位系统 (4周)
```yaml
目标: 实现骑手管理和实时定位追踪
关键交付:
- 骑手注册认证
- 实时位置上传
- 订单智能分配
- 配送路径优化
- 绩效统计系统
```
**实时定位系统:**
```python
from fastapi import WebSocket, WebSocketDisconnect
from typing import List, Dict
import asyncio
import json
class LocationManager:
def __init__(self):
self.active_connections: Dict[int, WebSocket] = {}
self.rider_locations: Dict[int, Dict] = {}
async def connect_rider(self, websocket: WebSocket, rider_id: int):
await websocket.accept()
self.active_connections[rider_id] = websocket
async def disconnect_rider(self, rider_id: int):
if rider_id in self.active_connections:
del self.active_connections[rider_id]
if rider_id in self.rider_locations:
del self.rider_locations[rider_id]
async def update_location(self, rider_id: int, location: Dict):
self.rider_locations[rider_id] = {
**location,
'updated_at': datetime.utcnow()
}
# 广播给相关订单用户
await self.broadcast_location_update(rider_id, location)
async def get_nearby_riders(self, lat: float, lng: float, radius: float = 5.0):
"""获取附近骑手"""
nearby_riders = []
for rider_id, location_data in self.rider_locations.items():
rider_lat = location_data['latitude']
rider_lng = location_data['longitude']
distance = self.calculate_distance(lat, lng, rider_lat, rider_lng)
if distance <= radius:
nearby_riders.append({
'rider_id': rider_id,
'distance': distance,
'location': location_data
})
return sorted(nearby_riders, key=lambda x: x['distance'])
def calculate_distance(self, lat1: float, lng1: float, lat2: float, lng2: float) -> float:
"""计算两点间距离（公里）"""
from math import radians, cos, sin, asin, sqrt
# 转换为弧度
lat1, lng1, lat2, lng2 = map(radians, [lat1, lng1, lat2, lng2])
# haversine公式
dlat = lat2 - lat1
dlng = lng2 - lng1
a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlng/2)**2
c = 2 * asin(sqrt(a))
r = 6371 # 地球半径（公里）
return c * r
# WebSocket定位上传
@router.websocket("/ws/rider/{rider_id}/location")
async def location_websocket(websocket: WebSocket, rider_id: int):
await location_manager.connect_rider(websocket, rider_id)
try:
while True:
data = await websocket.receive_text()
location_data = json.loads(data)
# 验证位置数据格式
if 'latitude' in location_data and 'longitude' in location_data:
await location_manager.update_location(rider_id, location_data)
except WebSocketDisconnect:
await location_manager.disconnect_rider(rider_id)
```
**智能订单分配算法:**
```python
from typing import List, Tuple
import asyncio
class OrderDispatcher:
def __init__(self, location_manager: LocationManager):
self.location_manager = location_manager
async def dispatch_order(self, order_id: int, store_location: Dict) -> Optional[int]:
"""智能分配订单给骑手"""
# 获取订单详情
order = await self.get_order(order_id)
if not order:
return None
# 获取附近骑手
nearby_riders = await self.location_manager.get_nearby_riders(
store_location['latitude'],
store_location['longitude'],
radius=10.0
)
# 筛选可用骑手
available_riders = await self.filter_available_riders(nearby_riders)
if not available_riders:
return None
# 综合评分算法
best_rider = await self.select_best_rider(available_riders, order)
if best_rider:
# 分配订单
await self.assign_order_to_rider(order_id, best_rider['rider_id'])
return best_rider['rider_id']
return None
async def filter_available_riders(self, riders: List[Dict]) -> List[Dict]:
"""筛选可用骑手"""
available_riders = []
for rider in riders:
# 检查骑手工作状态
if rider['work_status'] != 1: # 1为在线
continue
# 检查当前订单数量
if rider['current_orders'] >= 3: # 最多同时接3单
continue
# 检查评分
if rider['rating'] < 4.0: # 评分低于4.0不分配
continue
available_riders.append(rider)
return available_riders
async def select_best_rider(self, riders: List[Dict], order: Dict) -> Optional[Dict]:
"""选择最佳骑手"""
if not riders:
return None
def calculate_score(rider: Dict) -> float:
distance_score = 10 / (rider['distance'] + 0.1) # 距离越近分数越高
rating_score = rider['rating'] * 2 # 评分权重
orders_score = rider['completed_orders'] * 0.01 # 经验权重
on_time_score = rider['on_time_rate'] * 0.1 # 准时率权重
return distance_score + rating_score + orders_score + on_time_score
# 计算每个骑手的综合评分
scored_riders = []
for rider in riders:
score = calculate_score(rider)
scored_riders.append({**rider, 'dispatch_score': score})
# 返回评分最高的骑手
return max(scored_riders, key=lambda x: x['dispatch_score'])
```
### 3.2 配送系统与路径优化 (3周)
```python
from typing import List, Tuple
import asyncio
import requests
class DeliveryService:
def __init__(self):
self.map_service = MapService() # 地图服务（高德/百度）
async def plan_delivery_route(self, orders: List[Dict], start_location: Dict) -> Dict:
"""规划最优配送路径"""
if not orders:
return {"route": [], "total_distance": 0, "estimated_time": 0}
# 构建途经点
waypoints = [
(order['delivery_address']['latitude'], order['delivery_address']['longitude'])
for order in orders
]
# 调用地图API规划路径
route_result = await self.map_service.get_optimal_route(
start=(start_location['latitude'], start_location['longitude']),
waypoints=waypoints
)
return {
"route": route_result['route'],
"total_distance": route_result['distance'],
"estimated_time": route_result['duration'],
"orders_sequence": self.optimize_order_sequence(orders, route_result['route'])
}
def optimize_order_sequence(self, orders: List[Dict], route: List[Dict]) -> List[int]:
"""根据路径优化订单配送顺序"""
# 简单实现：按照路径顺序排列订单
# 实际可考虑更多因素：配送时间窗口、优先级等
order_sequence = []
for point in route[1:]: # 排除起点
# 找到最近的订单
nearest_order = min(
orders,
key=lambda o: self.calculate_distance(
point['latitude'], point['longitude'],
o['delivery_address']['latitude'],
o['delivery_address']['longitude']
)
)
if nearest_order not in order_sequence:
order_sequence.append(nearest_order)
return [order['id'] for order in order_sequence]
async def update_delivery_status(self, order_id: int, status: str, location: Dict = None):
"""更新配送状态"""
# 更新订单状态
await self.order_service.update_status(order_id, status)
# 更新骑手位置
if location:
await self.location_manager.update_location(
self.get_rider_id_by_order(order_id),
location
)
# 发送通知给用户
if status in ['picking', 'delivering', 'delivered']:
await self.notification_service.send_delivery_update(
order_id, status
)
class MapService:
def __init__(self):
self.api_key = "your_map_api_key"
self.base_url = "https://restapi.amap.com/v3"
async def get_optimal_route(self, start: Tuple[float, float], waypoints: List[Tuple[float, float]]) -> Dict:
"""调用高德地图API获取最优路径"""
url = f"{self.base_url}/direction/driving"
params = {
'key': self.api_key,
'origin': f"{start[1]},{start[0]}", # 经度,纬度
'destination': f"{waypoints[-1][1]},{waypoints[-1][0]}",
'waypoints': ';'.join([f"{wp[1]},{wp[0]}" for wp in waypoints[:-1]]),
'strategy': 1, # 距离优先策略
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
data = await response.json()
if data.get('status') == '1' and data.get('count') != '0':
route = data['route']['paths'][0]
return {
'distance': int(route['distance']) / 1000, # 转换为公里
'duration': int(route['duration']) // 60, # 转换为分钟
'route': self.parse_route_steps(route['steps'])
}
else:
raise Exception("Failed to get route from map API")
def parse_route_steps(self, steps: List[Dict]) -> List[Dict]:
"""解析路径步骤"""
route = []
for step in steps:
instruction = step['instruction']
distance = int(step['distance']) / 1000 # 转换为公里
duration = int(step['duration']) // 60 # 转换为分钟
# 解析坐标点
polyline = step['polyline']
points = self.decode_polyline(polyline)
route.append({
'instruction': instruction,
'distance': distance,
'duration': duration,
'coordinates': points
})
return route
```
---
## 💎 第四阶段：系统优化与完善 (10-12个月)
### 4.1 性能优化与监控 (3周)
```yaml
目标: 全面优化系统性能和建立监控体系
关键交付:
- 数据库查询优化
- Redis缓存策略
- API响应时间优化
- 全链路监控
- 压力测试与调优
```
**缓存策略设计:**
```python
from functools import wraps
from typing import Any, Optional
import redis
import json
import hashlib
class CacheManager:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.default_ttl = 3600 # 1小时
def cache(self, key_prefix: str, ttl: int = None):
"""缓存装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 生成缓存键
cache_key = self.generate_cache_key(key_prefix, *args, **kwargs)
# 尝试从缓存获取
cached_result = await self.get(cache_key)
if cached_result is not None:
return cached_result
# 执行原函数
result = await func(*args, **kwargs)
# 存储到缓存
await self.set(cache_key, result, ttl or self.default_ttl)
return result
return wrapper
return decorator
def generate_cache_key(self, prefix: str, *args, **kwargs) -> str:
"""生成缓存键"""
key_data = {
'args': args,
'kwargs': sorted(kwargs.items())
}
key_hash = hashlib.md5(json.dumps(key_data, sort_keys=True).encode()).hexdigest()
return f"{prefix}:{key_hash}"
async def get(self, key: str) -> Optional[Any]:
"""获取缓存"""
try:
data = await self.redis.get(key)
if data:
return json.loads(data)
except Exception as e:
print(f"Cache get error: {e}")
return None
async def set(self, key: str, value: Any, ttl: int) -> bool:
"""设置缓存"""
try:
data = json.dumps(value, default=str)
await self.redis.setex(key, ttl, data)
return True
except Exception as e:
print(f"Cache set error: {e}")
return False
# 使用示例
cache_manager = CacheManager(redis_client)
@cache_manager.cache("order_details", ttl=1800) # 30分钟
async def get_order_details(order_id: int):
"""获取订单详情（带缓存）"""
order = await order_repository.get_by_id(order_id)
if order:
# 加载关联数据
order.items = await order_item_repository.get_by_order_id(order_id)
order.payments = await payment_repository.get_by_order_id(order_id)
return order
@cache_manager.cache("rider_stats", ttl=300) # 5分钟
async def get_rider_statistics(rider_id: int, date_range: str):
"""获取骑手统计数据（带缓存）"""
return await rider_repository.get_statistics(rider_id, date_range)
```
**性能监控:**
```python
from prometheus_client import Counter, Histogram, Gauge
import time
from functools import wraps
# Prometheus指标
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 'HTTP request duration')
ACTIVE_CONNECTIONS = Gauge('active_connections', 'Active connections')
def monitor_requests(func):
"""请求监控装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = await func(*args, **kwargs)
status = 'success'
return result
except Exception as e:
status = 'error'
raise
finally:
duration = time.time() - start_time
REQUEST_DURATION.observe(duration)
REQUEST_COUNT.labels(
method=kwargs.get('method', 'unknown'),
endpoint=kwargs.get('endpoint', func.__name__),
status=status
).inc()
return wrapper
# 中间件监控
@app.middleware("http")
async def monitoring_middleware(request: Request, call_next):
start_time = time.time()
# 增加活跃连接数
ACTIVE_CONNECTIONS.inc()
try:
response = await call_next(request)
# 记录请求指标
duration = time.time() - start_time
REQUEST_DURATION.observe(duration)
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.url.path,
status='success'
).inc()
# 添加响应头
response.headers["X-Process-Time"] = str(duration)
return response
except Exception as e:
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.url.path,
status='error'
).inc()
raise
finally:
# 减少活跃连接数
ACTIVE_CONNECTIONS.dec()
```
### 4.2 Ant Design Pro Vue前端开发 (6-8周)
**前端项目结构:**
```
frontend/
├── src/
│ ├── components/ # 通用组件
│ │ ├── ProTable/
│ │ ├── ProForm/
│ │ └── Charts/
│ ├── pages/ # 页面组件
│ │ ├── order/
│ │ │ ├── list.vue
│ │ │ ├── detail.vue
│ │ │ └── create.vue
│ │ ├── rider/
│ │ ├── store/
│ │ └── dashboard/
│ ├── services/ # API服务
│ │ ├── order.ts
│ │ ├── rider.ts
│ │ └── user.ts
│ ├── stores/ # Pinia状态管理
│ │ ├── order.ts
│ │ ├── user.ts
│ │ └── app.ts
│ ├── utils/ # 工具函数
│ ├── types/ # TypeScript类型定义
│ └── layouts/ # 布局组件
├── public/
├── package.json
└── vite.config.ts
```
**核心页面组件示例:**
```vue
{{ getStatusText(record.status) }}
{{ getPaymentStatusText(record.payment_status) }}
查看
开始配送
```
---
## 🧪 测试策略与质量保证
### 测试框架设计
```python
# pytest配置
import pytest
import asyncio
from httpx import AsyncClient
from app.main import app
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
@pytest.fixture
async def test_user_token():
# 创建测试用户并返回JWT token
user_data = {
"phone": "13800138000",
"password": "test123456"
}
# ...创建用户逻辑
return "test_jwt_token"
# 订单API测试
@pytest.mark.asyncio
async def test_create_order(client: AsyncClient, test_user_token: str):
order_data = {
"store_id": 1,
"items": [
{"goods_id": 1, "quantity": 2, "price": 12.50},
{"goods_id": 2, "quantity": 1, "price": 8.00}
],
"delivery_address": {
"address": "测试地址",
"latitude": 39.9042,
"longitude": 116.4074,
"contact_phone": "13800138000"
}
}
response = await client.post(
"/api/v1/orders",
json=order_data,
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["status"] == 1
assert "order_id" in data["data"]
assert data["data"]["final_amount"] == 33.00
# 支付测试
@pytest.mark.asyncio
async def test_payment_process(client: AsyncClient, test_user_token: str):
# 首先创建订单
order_response = await test_create_order(client, test_user_token)
order_id = order_response.json()["data"]["order_id"]
# 测试支付
payment_data = {
"order_id": order_id,
"payment_method": "wechat",
"amount": 33.00
}
response = await client.post(
"/api/v1/payments",
json=payment_data,
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["status"] == 1
assert "payment_params" in data["data"]
```
---
## 📊 性能指标与监控
### 关键性能指标
```yaml
系统性能目标:
API响应时间: < 200ms (P95)
数据库查询时间: < 100ms (P95)
系统可用性: > 99.9%
并发用户数: 10,000+
订单处理能力: 1,000 TPS
业务监控指标:
订单创建成功率: > 99.5%
支付成功率: > 98%
配送准时率: > 95%
骑手接单响应时间: < 30秒
```
---
## 🔧 部署与运维
### Docker容器化
```dockerfile
# FastAPI后端Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
# Vue前端Dockerfile
FROM node:16-alpine as build
WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY . .
RUN npm run build
FROM nginx:alpine
COPY --from=build /app/dist /usr/share/nginx/html
COPY nginx.conf /etc/nginx/nginx.conf
EXPOSE 80
```
### Docker Compose配置
```yaml
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=mysql+asyncmy://user:password@db:3306/ecommerce
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs
db:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: rootpassword
MYSQL_DATABASE: ecommerce
MYSQL_USER: user
MYSQL_PASSWORD: password
volumes:
- mysql_data:/var/lib/mysql
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "3306:3306"
redis:
image: redis:6-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./ssl:/etc/nginx/ssl
depends_on:
- app
volumes:
mysql_data:
redis_data:
```
---
## ⚠️ 风险控制与应急预案
### 数据安全
- **数据加密**: 敏感数据AES加密存储
- **传输安全**: HTTPS/TLS 1.3加密
- **访问控制**: 细粒度权限管理
- **审计日志**: 全链路操作审计
### 系统稳定性
- **熔断机制**: 防止雪崩效应
- **限流控制**: API请求频率限制
- **降级策略**: 核心功能优先保证
- **备份方案**: 实时备份和快速恢复
### 迁移风险
- **灰度发布**: 逐步切换流量
- **监控告警**: 实时系统健康监控
- **回滚预案**: 快速回退到旧系统
- **数据校验**: 新旧系统数据一致性检查
---
## 📈 项目总结
### 预期收益
1. **性能提升**: API响应时间提升60%+，系统吞吐量提升300%+
2. **开发效率**: 现代化开发框架，代码质量和维护性大幅提升
3. **用户体验**: 实时功能、更好的UI交互、更快的页面加载
4. **运维效率**: 容器化部署、自动化运维、完善监控体系
5. **扩展性**: 微服务架构，支持业务快速迭代和横向扩展
### 技术债务清理
- ✅ 移除ThinkPHP 3.x过时框架
- ✅ 升级到现代化技术栈
- ✅ 重构复杂业务逻辑
- ✅ 建立完善测试体系
- ✅ 实施DevOps最佳实践
这个重构计划将彻底现代化电商平台的技术架构，为未来的业务发展奠定坚实基础。