Python架构演进与迁移

系统架构演进需要循序渐进，避免大规模重构带来的风险。

架构演进路径

单体到微服务演进阶段

单体架构 → 模块化单体 → 分布式单体 → 微服务

阶段1：单体架构
- 所有功能在一个应用中
- 共享数据库
- 部署简单但扩展困难

阶段2：模块化单体
- 按业务模块划分
- 模块边界清晰
- 共享数据库但模块隔离

阶段3：分布式单体
- 模块拆分为独立服务
- 服务间RPC调用
- 数据库仍共享

阶段4：微服务
- 服务完全独立
- 独立数据库
- 独立部署与扩展

演进决策因素

# 演进时机判断
def evaluate_architecture_evolution():
    factors = {
        'team_size': 5,          # 团队规模
        'user_scale': 'medium',  # 用户规模
        'feature_count': 20,     # 功能模块数
        'deployment_freq': 'daily',  # 部署频率
        'performance_req': 'high',   # 性能需求
    }

    # 决策逻辑
    if factors['team_size'] < 10:
        return "Stay with modular monolith"

    if factors['user_scale'] == 'small':
        return "Monolith is sufficient"

    if factors['deployment_freq'] == 'rare':
        return "No need to split yet"

    if factors['performance_req'] == 'high':
        return "Consider microservices for scaling"

    return "Gradual evolution recommended"

模块化单体

模块边界设计

# 项目结构
project/
├── modules/
│   ├── user/          # 用户模块
│   │   ├── api.py     # API接口
│   │   ├── service.py # 业务逻辑
│   │   ├── repository.py # 数据访问
│   │   ├── models.py  # 数据模型
│   │   └── __init__.py
│   ├── order/
│   │   ├── api.py
│   │   ├── service.py
│   │   └── ...
│   ├── product/
│   │   └── ...
│   └── payment/
│       └── ...
├── core/              # 共享核心
│   ├── config.py
│   ├── database.py
│   └── exceptions.py
└── main.py            # 应用入口

# 模块接口定义
# user/api.py
class UserAPI:
    def __init__(self, user_service):
        self.service = user_service

    def get_user(self, user_id):
        return self.service.get_user(user_id)

    def create_user(self, data):
        return self.service.create_user(data)

模块间通信

# 模块间通过服务接口通信，不直接访问其他模块内部

# order/service.py
class OrderService:
    def __init__(self, order_repo, user_api, product_api):
        self.repo = order_repo
        self.user_api = user_api    # 通过API访问用户模块
        self.product_api = product_api

    def create_order(self, user_id, product_ids):
        # 通过用户模块API验证用户
        user = self.user_api.get_user(user_id)
        if not user:
            raise ValueError("User not found")

        # 通过产品模块API获取产品
        products = [self.product_api.get_product(pid) for pid in product_ids]

        order = Order(user_id, products)
        return self.repo.save(order)

服务拆分策略

按业务边界拆分

# 拆分优先级判断
def determine_split_priority():
    # 优先拆分特征：
    # 1. 业务独立性强
    # 2. 变化频率高
    # 3. 性能瓶颈明显
    # 4. 团队独立维护

    modules = {
        'notification': {'independence': 'high', 'change_freq': 'medium'},
        'payment': {'independence': 'high', 'change_freq': 'low'},
        'user': {'independence': 'medium', 'change_freq': 'high'},
        'order': {'independence': 'low', 'change_freq': 'high'},
    }

    # 计算拆分优先级
    for name, attrs in modules.items():
        score = (
            attrs['independence'] == 'high' * 2 +
            attrs['change_freq'] == 'high'
        )

Strangler Fig模式

# 逐步替换旧系统

# 步骤1：在旧系统旁创建新服务
# 新服务处理部分流量

# 步骤2：路由层控制流量
class Router:
    def route_request(self, request):
        path = request.path

        # 新服务处理的路径
        new_paths = ['/api/v2/users', '/api/v2/products']

        if path in new_paths:
            return self.new_service.handle(request)
        else:
            return self.old_system.handle(request)

# 步骤3：逐步迁移功能
# 每次迁移一个模块，验证后继续

# 步骤4：删除旧系统代码
# 所有功能迁移完成后关闭旧系统

数据迁移策略

共享数据库过渡

# 阶段1：共享数据库
# 服务拆分但数据库共享

# user/service.py
class UserService:
    def __init__(self, db):
        self.db = db  # 共享数据库连接

    def get_user(self, user_id):
        return self.db.query(User).get(user_id)

# order/service.py
class OrderService:
    def __init__(self, db, user_service):
        self.db = db  # 共享同一个数据库
        self.user_service = user_service

数据库分离

# 阶段2：独立数据库

# user数据库
user_db = Database('postgresql://user_db')

# order数据库
order_db = Database('postgresql://order_db')

# 数据同步策略
class UserDataSync:
    def sync_to_order_service(self, user):
        # 同步用户数据到订单服务
        order_api.create_user_reference(user.id, user.name)

# 或使用事件驱动
class UserEventPublisher:
    def publish_user_created(self, user):
        event = {'type': 'user_created', 'data': user}
        message_queue.publish('user_events', event)

数据迁移流程

class DataMigration:
    def migrate_user_data(self):
        # 1. 创建新数据库表结构
        self.create_new_schema()

        # 2. 复制数据到新数据库
        self.copy_data_batch()

        # 3. 双写验证
        self.enable_dual_write()

        # 4. 切换读源
        self.switch_read_source()

        # 5. 关闭旧写
        self.stop_old_write()

        # 6. 清理旧数据
        self.cleanup_old_data()

    def copy_data_batch(self):
        batch_size = 1000
        offset = 0

        while True:
            users = self.old_db.query(User).offset(offset).limit(batch_size).all()
            if not users:
                break

            for user in users:
                self.new_db.insert(UserNew.from_old(user))

            offset += batch_size

服务通信演进

同步调用过渡

# 初期：HTTP同步调用
import requests

class OrderService:
    def get_user(self, user_id):
        # HTTP调用用户服务
        resp = requests.get(f'http://user-service/users/{user_id}')
        return resp.json()

# 问题：耦合度高，性能受限

异步消息演进

# 后期：消息队列异步通信
import pika

class OrderService:
    def __init__(self, mq_connection):
        self.channel = mq_connection.channel()
        self.channel.queue_declare('user_requests')

    def request_user_async(self, user_id, callback):
        # 发送请求消息
        self.channel.basic_publish(
            exchange='',
            routing_key='user_requests',
            body=json.dumps({'user_id': user_id, 'callback': callback})
        )

# 用户服务处理请求
class UserService:
    def listen_requests(self):
        self.channel.basic_consume(
            queue='user_requests',
            on_message_callback=self.handle_request
        )

迁移风险管理

渐进式迁移原则

MIGRATION_RULES = {
    'incremental': '每次只迁移一个小模块',
    'rollback_ready': '随时可以回滚到旧系统',
    'monitor': '迁移过程持续监控',
    'test': '每个阶段都有完整测试',
    'traffic_control': '控制新服务流量比例',
}

流量切换控制

class TrafficController:
    def __init__(self):
        self.new_service_ratio = 0.0  # 新服务流量比例

    def set_ratio(self, ratio):
        "设置新服务流量比例"
        self.new_service_ratio = ratio

    def route(self, request):
        if random.random() < self.new_service_ratio:
            return self.new_service.handle(request)
        else:
            return self.old_service.handle(request)

    def gradual_increase(self):
        "逐步增加新服务流量"
        for ratio in [0.1, 0.2, 0.5, 0.8, 1.0]:
            self.set_ratio(ratio)
            time.sleep(86400)  # 每天增加
            self.monitor_errors()

回滚机制

class RollbackManager:
    def __init__(self):
        self.snapshots = []

    def create_snapshot(self, version):
        "创建迁移快照"
        snapshot = {
            'version': version,
            'timestamp': datetime.now(),
            'data_state': self.capture_data_state(),
            'config': self.capture_config()
        }
        self.snapshots.append(snapshot)

    def rollback(self, version):
        "回滚到指定版本"
        snapshot = self.find_snapshot(version)
        self.restore_data_state(snapshot['data_state'])
        self.restore_config(snapshot['config'])

迁移检查清单

注意：架构演进要渐进式进行，避免一次性大规模重构，每个阶段都要有回滚方案。

要点总结

• 架构演进路径：单体→模块化单体→分布式单体→微服务
• 模块化单体：按业务划分模块、清晰边界、通过接口通信
• Strangler Fig模式：新系统旁建旧系统、逐步替换功能
• 数据迁移：共享数据库过渡→独立数据库→数据同步机制
• 流量控制：渐进式切换、监控验证、随时可回滚
• 迁移原则：小步前进、回滚准备、持续监控、完整测试

存放路径：articles/PYTHON/专家/架构与设计/架构演进与迁移.md