27.2 复杂系统设计与实现

复杂系统设计概述

Claude Code 可以帮助开发者设计和实现复杂的软件系统。通过系统化的设计方法和分阶段的实现策略，开发者可以利用 Claude Code 构建可扩展、可维护的复杂系统。

系统架构设计

1. 微服务架构

示例：设计微服务架构

用户请求： "设计一个电商系统的微服务架构"

Claude Code 生成的架构设计：

系统架构

 
    │                        API Gateway                          │
    │  (负载均衡、认证、路由、限流、监控)                           │
    └─────────────────────────────────────────────────────────────┘
    │
    ┌─────────────────────┼─────────────────────┐
    │                     │                     │
    ┌───────▼────────┐  ┌────────▼────────┐  ┌────────▼────────┐
    │  User Service  │  │ Product Service │  │ Order Service   │
    │  - 用户管理     │  │  - 商品管理      │  │  - 订单管理      │
    │  - 认证授权     │  │  - 库存管理      │  │  - 支付处理      │
    │  - 个人资料     │  │  - 分类管理      │  │  - 物流跟踪      │
    └────────────────┘  └─────────────────┘  └─────────────────┘
    │                     │                     │
    └─────────────────────┼─────────────────────┘
    │
    ┌─────────────────────┼─────────────────────┐
    │                     │                     │
    ┌───────▼────────┐  ┌────────▼────────┐  ┌────────▼────────┐
    │ Payment Service│  │ Inventory      │  │ Notification    │
    │  - 支付处理     │  │ Service        │  │ Service         │
    │  - 退款处理     │  │  - 库存同步     │  │  - 邮件通知      │
    │  - 交易记录     │  │  - 预警管理     │  │  - 短信通知      │
    └────────────────┘  └─────────────────┘  └─────────────────┘
    │
    ┌─────────▼─────────┐
    │   Message Broker  │
    │  (Kafka/RabbitMQ) │
    └───────────────────┘
 
    ## 服务定义
 
```python
    ### User Service (用户服务)
 
```python
 
    # services/user_service/app.py
 
    from flask import Flask, request, jsonify
    from flask_sqlalchemy import SQLAlchemy
    from flask_jwt_extended import JWTManager, create_access_token
    import bcrypt
 
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://localhost/users_db'
    app.config['JWT_SECRET_KEY'] = 'your-secret-key'
 
    db = SQLAlchemy(app)
    jwt = JWTManager(app)
 
    class User(db.Model):
        """用户模型"""
        __tablename__ = 'users'
 
        id = db.Column(db.Integer, primary_key=True)
        email = db.Column(db.String(120), unique=True, nullable=False)
        password_hash = db.Column(db.String(255), nullable=False)
        name = db.Column(db.String(100), nullable=False)
        phone = db.Column(db.String(20))
        created_at = db.Column(db.DateTime, server_default=db.func.now())
 
        def set_password(self, password):
            """设置密码"""
            self.password_hash = bcrypt.hashpw(
                password.encode('utf-8'),
                bcrypt.gensalt()
            ).decode('utf-8')
 
        def check_password(self, password):
            """验证密码"""
            return bcrypt.checkpw(
                password.encode('utf-8'),
                self.password_hash.encode('utf-8')
            )
 
    @app.route('/register', methods=['POST'])
    def register():
        """用户注册"""
        data = request.get_json()
 
        user = User(
            email=data['email'],
            name=data['name'],
            phone=data.get('phone')
        )
        user.set_password(data['password'])
 
        db.session.add(user)
        db.session.commit()
 
        return jsonify({'user_id': user.id}), 201
 
    @app.route('/login', methods=['POST'])
    def login():
        """用户登录"""
        data = request.get_json()
 
        user = User.query.filter_by(email=data['email']).first()
 
        if not user or not user.check_password(data['password']):
            return jsonify({'error': 'Invalid credentials'}), 401
 
        access_token = create_access_token(identity=user.id)
 
        return jsonify({
            'access_token': access_token,
            'user': {
                'id': user.id,
                'email': user.email,
                'name': user.name
            }
        })
 
    @app.route('/users/<int:user_id>', methods=['GET'])
    def get_user(user_id):
        """获取用户信息"""
        user = User.query.get_or_404(user_id)
        return jsonify({
            'id': user.id,
            'email': user.email,
            'name': user.name,
            'phone': user.phone
        })
 
    ### Product Service (商品服务)
 
    ```python
 
    # services/product_service/app.py
```python
    from flask import Flask, request, jsonify
    from flask_sqlalchemy import SQLAlchemy
    from flask_migrate import Migrate
    from datetime import datetime
 
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://localhost/products_db'
 
    db = SQLAlchemy(app)
    migrate = Migrate(app, db)
 
    class Product(db.Model):

"""商品模型""" tablename = 'products'

id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(200), nullable=False) description = db.Column(db.Text) price = db.Column(db.Float, nullable=False) stock = db.Column(db.Integer, default=0) category_id = db.Column(db.Integer, db.ForeignKey('categories.id')) created_at = db.Column(db.DateTime, server_default=db.func.now()) updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())

category = db.relationship('Category', backref='products')

    class Category(db.Model):

"""分类模型""" tablename = 'categories'

id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(100), nullable=False) parent_id = db.Column(db.Integer, db.ForeignKey('categories.id'))

parent = db.relationship('Category', remote_side=[id], backref='children')

@app.route('/products', methods=['POST'])

    def create_product():

"""创建商品""" data = request.get_json()

product = Product( name=data['name'], description=data.get('description'), price=data['price'], stock=data.get('stock', 0), category_id=data.get('category_id') )

db.session.add(product) db.session.commit()

    return jsonify({'product_id': product.id}), 201
 
    @app.route('/products/<int:product_id>', methods=['GET'])
    def get_product(product_id):

"""获取商品详情""" product = Product.query.get_or_404(product_id)

    return jsonify({
    'id': product.id,
    'name': product.name,
    'description': product.description,
    'price': product.price,
    'stock': product.stock,
    'category': product.category.name if product.category else None
    })
 
    @app.route('/products/<int:product_id>/stock', methods=['PUT'])
    def update_stock(product_id):

"""更新库存""" data = request.get_json()

product = Product.query.get_or_404(product_id) product.stock = data['stock']

db.session.commit()

    return jsonify({'stock': product.stock})

Order Service (订单服务)

 
    # services/order_service/app.py
 
    from flask import Flask, request, jsonify
    from flask_sqlalchemy import SQLAlchemy
    from datetime import datetime
    import requests
 
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://localhost/orders_db'
 
    db = SQLAlchemy(app)
 
    class Order(db.Model):
        """订单模型"""
        __tablename__ = 'orders'
 
        id = db.Column(db.Integer, primary_key=True)
        user_id = db.Column(db.Integer, nullable=False)
        total_amount = db.Column(db.Float, nullable=False)
        status = db.Column(db.String(20), default='pending')
        created_at = db.Column(db.DateTime, server_default=db.func.now())
        updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())
 
    class OrderItem(db.Model):
        """订单项模型"""
        __tablename__ = 'order_items'
 
        id = db.Column(db.Integer, primary_key=True)
        order_id = db.Column(db.Integer, db.ForeignKey('orders.id'), nullable=False)
        product_id = db.Column(db.Integer, nullable=False)
        quantity = db.Column(db.Integer, nullable=False)
        price = db.Column(db.Float, nullable=False)
 
        order = db.relationship('Order', backref='items')
 
    @app.route('/orders', methods=['POST'])
    def create_order():
        """创建订单"""
        data = request.get_json()
 
        # 验证商品库存
 
        for item in data['items']:
            response = requests.get(
                f"http://product-service:5001/products/{item['product_id']}"
            )
            product = response.json()
 
            if product['stock'] < item['quantity']:
                return jsonify({
                    'error': f'Insufficient stock for product {item["product_id"]}'
                }), 400
 
        # 创建订单
 
        order = Order(
            user_id=data['user_id'],
            total_amount=0
        )
        db.session.add(order)
        db.session.flush()
 
        # 创建订单项
 
        total_amount = 0
        for item in data['items']:
            order_item = OrderItem(
                order_id=order.id,
                product_id=item['product_id'],
                quantity=item['quantity'],
                price=item['price']
            )
            db.session.add(order_item)
            total_amount += item['price'] * item['quantity']
 
        order.total_amount = total_amount
        db.session.commit()
 
        # 扣减库存
 
        for item in data['items']:
            response = requests.get(
                f"http://product-service:5001/products/{item['product_id']}"
            )
            product = response.json()
            new_stock = product['stock'] - item['quantity']
 
            requests.put(
                f"http://product-service:5001/products/{item['product_id']}/stock",
                json={'stock': new_stock}
            )
 
        return jsonify({'order_id': order.id}), 201
 
    @app.route('/orders/<int:order_id>', methods=['GET'])
    def get_order(order_id):
        """获取订单详情"""
        order = Order.query.get_or_404(order_id)
 
        return jsonify({
            'id': order.id,
            'user_id': order.user_id,
            'total_amount': order.total_amount,
            'status': order.status,
            'items': [
                {
                    'product_id': item.product_id,
                    'quantity': item.quantity,
                    'price': item.price
                }
                for item in order.items
            ]
        })
 
    ## API Gateway 实现
 
```python
 
    # api_gateway/app.py
```python
    from flask import Flask, request, jsonify
    import requests
    import jwt
    from functools import wraps
 
    app = Flask(__name__)
 
    SERVICES = {
        'user': 'http://user-service:5000',
        'product': 'http://product-service:5001',
        'order': 'http://order-service:5002'
    }
 
    def authenticate(f):

"""认证装饰器""" @wraps(f)

        def decorated_function(*args, **kwargs):
            token = request.headers.get('Authorization')
 
            if not token:
                return jsonify({'error': 'Missing token'}), 401
 
            try:
                payload = jwt.decode(token.split()[1], 'your-secret-key', algorithms=['HS256'])
                request.user_id = payload['user_id']
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Invalid token'}), 401
 
            return f(*args, **kwargs)
 
        return decorated_function

    @app.route('/<service>/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])

    def proxy(service, path):

"""代理请求到对应的服务"""

        if service not in SERVICES:
            return jsonify({'error': 'Service not found'}), 404
 
        service_url = SERVICES[service]
        url = f"{service_url}/{path}"
 
        headers = {key: value for key, value in request.headers if key != 'Host'}
 
        response = requests.request(
            method=request.method,
            url=url,
            headers=headers,
            json=request.get_json(),
            params=request.args
        )
 
        return jsonify(response.json()), response.status_code
 
    @app.route('/health')
    def health():

"""健康检查"""

        return jsonify({'status': 'healthy'})

 
### 2. 事件驱动架构
 
````python
`python
 
# 示例：设计事件驱动架构
 
用户请求：
"设计一个基于事件驱动的订单处理系统"
 
Claude Code 生成的架构设计：
 
## 系统架构
 
```> ┌──────────────┐
 
│   订单服务    │
│  (Order      │
│   Service)   │
└──────┬───────┘
│ 发布事件
▼
┌─────────────────────────────────────────────────────────────┐
│                    Message Broker                           │
│                    (Kafka/RabbitMQ)                         │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ 订单创建  │  │ 订单支付  │  │ 订单发货  │  │ 订单完成  │  │
│  │ 事件     │  │ 事件     │  │ 事件     │  │ 事件     │  │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘  │
└─────────────────────────────────────────────────────────────┘
│              │              │              │
▼              ▼              ▼              ▼
┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────┐
│ 库存服务  │   │ 支付服务  │   │ 物流服务  │   │ 通知服务  │
│ Inventory│   │ Payment  │   │ Shipping │   │ Notify   │
│ Service  │   │ Service  │   │ Service  │   │ Service  │
└──────────┘   └──────────┘   └──────────┘   └──────────┘
 
## 事件定义
 
# events/order_events.py
 
```python
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Any
import json
@dataclass
class OrderEvent:

"""订单事件基类""" event_type: str order_id: int timestamp: datetime data: Dict[str, Any]

    def to_dict(self) -> dict:

"""转换为字典"""

    return &#123;
    'event_type': self.event_type,
    'order_id': self.order_id,
    'timestamp': self.timestamp.isoformat(),
    'data': self.data
    &#125;
    def to_json(self) -> str:
    """转换为 JSON"""
    return json.dumps(self.to_dict())
    @dataclass
    class OrderCreatedEvent(OrderEvent):

"""订单创建事件"""

    def __init__(self, order_id: int, user_id: int, total_amount: float, items: list):
    super().__init__(
    event_type='order.created',
    order_id=order_id,
    timestamp=datetime.utcnow(),
    data=&#123;
    'user_id': user_id,
    'total_amount': total_amount,
    'items': items
    &#125;
    )
    @dataclass
    class OrderPaidEvent(OrderEvent):

"""订单支付事件"""

    def __init__(self, order_id: int, payment_method: str, transaction_id: str):
    super().__init__(
    event_type='order.paid',
    order_id=order_id,
    timestamp=datetime.utcnow(),
    data=&#123;
    'payment_method': payment_method,
    'transaction_id': transaction_id
    &#125;
    )
    @dataclass
    class OrderShippedEvent(OrderEvent):

"""订单发货事件"""

    def __init__(self, order_id: int, tracking_number: str, shipping_address: dict):
    super().__init__(
    event_type='order.shipped',
    order_id=order_id,
    timestamp=datetime.utcnow(),
    data=&#123;
    'tracking_number': tracking_number,
    'shipping_address': shipping_address
    &#125;
    )
    @dataclass
    class OrderCompletedEvent(OrderEvent):

"""订单完成事件"""

    def __init__(self, order_id: int, delivery_time: datetime):
    super().__init__(
    event_type='order.completed',
    order_id=order_id,
    timestamp=datetime.utcnow(),
    data=&#123;
    'delivery_time': delivery_time.isoformat()
    &#125;
    )

事件发布者

 
    # services/order_service/event_publisher.py
    import pika
    import json
    from events.order_events import OrderEvent
 
    class EventPublisher:
        """事件发布者"""
 
        def __init__(self, rabbitmq_url: str = 'amqp://localhost'):
            self.connection = pika.BlockingConnection(pika.URLParameters(rabbitmq_url))
            self.channel = self.connection.channel()
 
            # 声明交换机
            self.channel.exchange_declare(
                exchange='order_events',
                exchange_type='topic'
            )
 
        def publish(self, event: OrderEvent, routing_key: str):
            """发布事件"""
            self.channel.basic_publish(
                exchange='order_events',
                routing_key=routing_key,
                body=event.to_json(),
                properties=pika.BasicProperties(
                    delivery_mode=2,  # 持久化
                )
            )
 
        def close(self):
            """关闭连接"""
            self.connection.close()
 
    ## 事件消费者
 
    # services/inventory_service/event_consumer.py
    import pika
    import json
    import requests
    class InventoryEventConsumer:
    """库存事件消费者"""
    def __init__(self, rabbitmq_url: str = 'amqp://localhost'):
    self.connection = pika.BlockingConnection(pika.URLParameters(rabbitmq_url))
    self.channel = self.connection.channel()
    # 声明交换机和队列
    self.channel.exchange_declare(
    exchange='order_events',
    exchange_type='topic'
    )
    self.channel.queue_declare(queue='inventory_queue', durable=True)
    self.channel.queue_bind(
    exchange='order_events',
    queue='inventory_queue',
    routing_key='order.created'
    )
    def handle_order_created(self, ch, method, properties, body):
    """处理订单创建事件"""
    event = json.loads(body)
    print(f"Received order.created event: &#123;event&#125;")
    # 扣减库存
    for item in event['data']['items']:
    response = requests.get(
    f"http://product-service:5001/products/&#123;item['product_id']&#125;"
    )
    product = response.json()
    new_stock = product['stock'] - item['quantity']
    requests.put(
    f"http://product-service:5001/products/&#123;item['product_id']&#125;/stock",
    json=&#123;'stock': new_stock&#125;
    )
    # 确认消息
    ch.basic_ack(delivery_tag=method.delivery_tag)
    def start_consuming(self):
    """开始消费消息"""
    self.channel.basic_qos(prefetch_count=1)
    self.channel.basic_consume(
    queue='inventory_queue',
    on_message_callback=self.handle_order_created
    )
    print('Inventory service started consuming messages...')
    self.channel.start_consuming()
    def close(self):
    """关闭连接"""
    self.connection.close()
 
    ## 事件驱动订单服务
 
```python
 
    # services/order_service/app.py
```python
    from flask import Flask, request, jsonify
    from flask_sqlalchemy import SQLAlchemy
    from datetime import datetime
    from event_publisher import EventPublisher
    from events.order_events import OrderCreatedEvent
 
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://localhost/orders_db'
 
    db = SQLAlchemy(app)
    event_publisher = EventPublisher()
 
    class Order(db.Model):

"""订单模型""" tablename = 'orders'

id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, nullable=False) total_amount = db.Column(db.Float, nullable=False) status = db.Column(db.String(20), default='pending') created_at = db.Column(db.DateTime, server_default=db.func.now()) updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())

    class OrderItem(db.Model):

"""订单项模型""" tablename = 'order_items'

id = db.Column(db.Integer, primary_key=True) order_id = db.Column(db.Integer, db.ForeignKey('orders.id'), nullable=False) product_id = db.Column(db.Integer, nullable=False) quantity = db.Column(db.Integer, nullable=False) price = db.Column(db.Float, nullable=False)

order = db.relationship('Order', backref='items')

@app.route('/orders', methods=['POST'])

    def create_order():

"""创建订单""" data = request.get_json()

创建订单

order = Order( user_id=data['user_id'], total_amount=0 ) db.session.add(order) db.session.flush()

创建订单项

total_amount = 0 items = []

        for item in data['items']:
            order_item = OrderItem(
                order_id=order.id,
                product_id=item['product_id'],
                quantity=item['quantity'],
                price=item['price']
            )
            db.session.add(order_item)
            total_amount += item['price'] * item['quantity']
            items.append(&#123;
                'product_id': item['product_id'],
                'quantity': item['quantity'],
                'price': item['price']
            &#125;)
 
        order.total_amount = total_amount
        db.session.commit()

发布订单创建事件

event = OrderCreatedEvent( order_id=order.id, user_id=data['user_id'], total_amount=total_amount, items=items )

        event_publisher.publish(event, 'order.created')
 
        return jsonify(&#123;'order_id': order.id&#125;), 201
 
    if __name__ == '__main__':
        app.run(host='0.0.0.0', port=5002)