后端系统架构设计：从单体到微服务的演进之路

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 单体应用的典型问题

# 问题1: 代码耦合严重
# 一个请求的处理流程涉及多个模块
class OrderService:
    def create_order(self, user_id, items):
        # 直接依赖多个模块
        user = UserService().get_user(user_id)
        inventory = InventoryService().check_stock(items)
        payment = PaymentService().process_payment(items)
        shipping = ShippingService().calculate_shipping(user.address)
        notification = NotificationService().send_confirmation(user.email)

        # 如果任何一个模块出错，整个订单创建失败
        return Order(user=user, items=items, payment=payment)

# 问题2: 难以独立扩展
# 当订单服务压力大时，必须整体扩展
# 无法针对特定瓶颈服务单独扩容

# 问题3: 技术栈锁定
# 整个应用必须使用相同的语言和框架
# 无法为新服务选择更适合的技术

# 问题4: 部署风险高
# 任何小的修改都需要重新部署整个应用
# 一处bug可能影响整个系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# ========== 服务拆分原则 ==========

# 1. 单一职责原则
# 每个服务专注于一个业务领域
class UserService:
    """用户服务 - 只负责用户相关的业务"""
    def create_user(self, data): pass
    def get_user(self, user_id): pass
    def update_user(self, user_id, data): pass

class OrderService:
    """订单服务 - 只负责订单相关的业务"""
    def create_order(self, data): pass
    def get_order(self, order_id): pass
    def cancel_order(self, order_id): pass

# 2. 限界上下文原则
# 按照业务领域边界拆分服务
# DDD (Domain-Driven Design) 战术模式

# 3. 数据独立性原则
# 每个服务拥有独立的数据库
class UserDatabase:
    """用户服务的数据库"""
    def __init__(self):
        self.db = PostgreSQL('users_db')

class OrderDatabase:
    """订单服务的数据库"""
    def __init__(self):
        self.db = MongoDB('orders_db')

# 4. API网关原则
# 统一入口，路由转发
class APIGateway:
    """API网关 - 服务统一入口"""

    def __init__(self):
        self.routes = {
            '/api/users/*': UserService(),
            '/api/orders/*': OrderService(),
            '/api/products/*': ProductService(),
            '/api/payments/*': PaymentService()
        }

    def route(self, request):
        # 路由匹配
        for pattern, service in self.routes.items():
            if request.path.match(pattern):
                return service.handle(request)

        # 聚合多个服务的响应
        if request.path == '/api/dashboard':
            return self.aggregate_dashboard(request)

    def aggregate_dashboard(self, request):
        """聚合多个服务的数据"""
        user_data = self.call_service('/api/users/me', request)
        order_data = self.call_service('/api/orders/recent', request)
        notification_data = self.call_service('/api/notifications', request)

        return {
            'user': user_data,
            'orders': order_data,
            'notifications': notification_data
        }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

# ========== 同步通信: REST/gRPC ==========

import requests
from typing import Protocol

# REST API调用
class OrderClient:
    """订单服务客户端"""

    BASE_URL = "http://order-service:8080"

    def create_order(self, order_data: dict) -> dict:
        response = requests.post(
            f"{self.BASE_URL}/api/orders",
            json=order_data,
            timeout=5  # 超时控制
        )
        response.raise_for_status()
        return response.json()

    def get_order(self, order_id: str) -> dict:
        response = requests.get(
            f"{self.BASE_URL}/api/orders/{order_id}",
            timeout=3
        )
        response.raise_for_status()
        return response.json()

# gRPC调用 (性能更高)
import grpc
from generated import order_pb2, order_pb2_grpc

class OrderGRPCClient:
    """订单服务gRPC客户端"""

    def __init__(self):
        self.channel = grpc.insecure_channel('order-service:9090')
        self.stub = order_pb2_grpc.OrderServiceStub(self.channel)

    def create_order(self, order_data: dict) -> order_pb2.OrderResponse:
        request = order_pb2.CreateOrderRequest(
            user_id=order_data['user_id'],
            items=[
                order_pb2.OrderItem(
                    product_id=item['product_id'],
                    quantity=item['quantity']
                )
                for item in order_data['items']
            ]
        )
        return self.stub.CreateOrder(request, timeout=5)

# ========== 异步通信: 消息队列 ==========

import asyncio
from aio_pika import connect, Message

class EventBus:
    """事件总线 - 异步消息传递"""

    def __init__(self, amqp_url: str):
        self.connection = None
        self.channel = None
        self.amqp_url = amqp_url

    async def connect(self):
        """建立连接"""
        self.connection = await connect(self.amqp_url)
        self.channel = await self.connection.channel()

        # 声明交换机
        await self.channel.declare_exchange(
            'domain-events',
            'topic',
            durable=True
        )

    async def publish(self, event_type: str, event_data: dict):
        """发布事件"""
        exchange = await self.get_exchange()

        message = Message(
            json.dumps(event_data).encode(),
            content_type='application/json',
            delivery_mode=2  # 持久化
        )

        await exchange.publish(
            message,
            routing_key=event_type
        )

    async def subscribe(self, event_pattern: str, handler):
        """订阅事件"""
        exchange = await self.get_exchange()

        # 声明队列
        queue = await self.channel.declare_queue(
            f'{event_pattern}-queue',
            durable=True
        )

        # 绑定交换机
        await queue.bind(exchange, routing_key=event_pattern)

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                try:
                    event_data = json.loads(message.body.decode())
                    await handler(event_data)
                    await message.ack()
                except Exception as e:
                    await message.nack()

# 事件驱动架构示例
class OrderCreatedEvent:
    """订单创建事件"""

    def __init__(self, order_id, user_id, items):
        self.event_type = 'order.created'
        self.data = {
            'order_id': order_id,
            'user_id': user_id,
            'items': items,
            'timestamp': datetime.now().isoformat()
        }

# 订单服务发布事件
async def create_order_with_event(order_data):
    # 创建订单
    order = await order_repository.create(order_data)

    # 发布事件
    event_bus = EventBus()
    await event_bus.publish(
        'order.created',
        OrderCreatedEvent(
            order.id,
            order.user_id,
            order.items
        ).data
    )

    return order

# 库存服务监听事件
async def handle_order_created(event_data):
    """处理订单创建事件 - 扣减库存"""
    order_id = event_data['order_id']
    items = event_data['items']

    for item in items:
        await inventory_service.deduct_stock(
            item['product_id'],
            item['quantity']
        )

    await event_bus.publish(
        'inventory.deducted',
        {'order_id': order_id, 'status': 'completed'}
    )

# 通知服务监听事件
async def handle_inventory_deducted(event_data):
    """处理库存扣减完成事件 - 发送通知"""
    order_id = event_data['order_id']

    order = await order_repository.get(order_id)
    await notification_service.send_order_confirmation(
        order.user_email,
        order_id
    )

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

# ========== 服务注册中心 ==========

import asyncio
from typing import Dict, List, Optional
from datetime import datetime, timedelta

class ServiceInstance:
    """服务实例"""

    def __init__(self, service_id: str, address: str, port: int):
        self.service_id = service_id
        self.address = address
        self.port = port
        self.last_heartbeat = datetime.now()

    @property
    def url(self) -> str:
        return f"http://{self.address}:{self.port}"

    def is_alive(self, timeout: int = 10) -> bool:
        """检查实例是否存活"""
        return (datetime.now() - self.last_heartbeat).seconds < timeout

class ServiceRegistry:
    """服务注册中心"""

    def __init__(self):
        self.services: Dict[str, List[ServiceInstance]] = {}

    def register(self, service_name: str, instance: ServiceInstance):
        """注册服务实例"""
        if service_name not in self.services:
            self.services[service_name] = []

        # 检查是否已存在
        existing = next(
            (i for i in self.services[service_name]
             if i.service_id == instance.service_id),
            None
        )

        if existing:
            # 更新心跳时间
            existing.last_heartbeat = datetime.now()
        else:
            # 新注册
            self.services[service_name].append(instance)

        print(f"Registered: {service_name} - {instance.url}")

    def deregister(self, service_name: str, service_id: str):
        """注销服务实例"""
        if service_name in self.services:
            self.services[service_name] = [
                i for i in self.services[service_name]
                if i.service_id != service_id
            ]

    def discover(self, service_name: str) -> Optional[ServiceInstance]:
        """服务发现 - 负载均衡"""
        if service_name not in self.services:
            return None

        # 过滤掉失效的实例
        alive_instances = [
            i for i in self.services[service_name]
            if i.is_alive()
        ]

        if not alive_instances:
            return None

        # 轮询负载均衡
        return alive_instances[hash(service_name) % len(alive_instances)]

        # 或者使用随机选择
        # return random.choice(alive_instances)

    def heartbeat(self, service_name: str, service_id: str):
        """接收心跳"""
        if service_name in self.services:
            for instance in self.services[service_name]:
                if instance.service_id == service_id:
                    instance.last_heartbeat = datetime.now()

# ========== 服务客户端 ==========
class ServiceClient:
    """服务客户端 - 带服务发现"""

    def __init__(self, registry: ServiceRegistry):
        self.registry = registry
        self.cache = {}  # 缓存服务地址

    async def call(self, service_name: str, endpoint: str, **kwargs):
        """调用服务"""
        # 从缓存或注册中心获取服务地址
        instance = self.cache.get(service_name)

        if not instance or not instance.is_alive():
            instance = self.registry.discover(service_name)
            if not instance:
                raise ServiceUnavailableException(f"Service {service_name} not found")

            self.cache[service_name] = instance

        # 构建请求URL
        url = f"{instance.url}{endpoint}"

        try:
            response = requests.post(url, json=kwargs, timeout=5)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            # 调用失败，清除缓存
            self.cache.pop(service_name, None)
            raise e

# ========== 使用示例 ==========
# 服务启动时注册
registry = ServiceRegistry()

async def start_service():
    service_instance = ServiceInstance(
        service_id=f"order-service-{os.getenv('INSTANCE_ID')}",
        address=os.getenv('SERVICE_ADDRESS'),
        port=int(os.getenv('SERVICE_PORT'))
    )

    registry.register('order-service', service_instance)

    # 定期发送心跳
    while True:
        await asyncio.sleep(5)
        registry.heartbeat('order-service', service_instance.service_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

# ========== 配置中心 ==========

import asyncio
import json
from typing import Any, Callable
from watchfiles import awatch

class ConfigCenter:
    """分布式配置中心"""

    def __init__(self, config_dir: str = './config'):
        self.config_dir = config_dir
        self.configs = {}
        self.watchers = {}  # config_key -> [callbacks]

    def load_config(self, service_name: str) -> dict:
        """加载服务配置"""
        config_file = f"{self.config_dir}/{service_name}.json"

        try:
            with open(config_file) as f:
                config = json.load(f)
                self.configs[service_name] = config
                return config
        except FileNotFoundError:
            return {}

    def get_config(self, service_name: str, key: str = None) -> Any:
        """获取配置"""
        config = self.configs.get(service_name, {})

        if key:
            return config.get(key)

        return config

    def watch_config(self, service_name: str, callback: Callable):
        """监听配置变化"""
        if service_name not in self.watchers:
            self.watchers[service_name] = []
        self.watchers[service_name].append(callback)

    async def watch_changes(self):
        """监听配置文件变化"""
        async for changes in awatch(self.config_dir):
            for change_type, config_path in changes:
                service_name = config_path.stem

                if change_type == Change.modified:
                    # 重新加载配置
                    old_config = self.configs.get(service_name, {})
                    new_config = self.load_config(service_name)

                    # 触发回调
                    if service_name in self.watchers:
                        for callback in self.watchers[service_name]:
                            await callback(old_config, new_config)

# ========== 使用示例 ==========
config_center = ConfigCenter()

# 加载配置
app_config = config_center.load_config('order-service')

# 监听配置变化
async def on_config_changed(old_config, new_config):
    """配置变化处理"""
    if old_config.get('log_level') != new_config.get('log_level'):
        # 重新配置日志级别
        logging.getLogger().setLevel(new_config['log_level'])

    if old_config.get('database') != new_config.get('database'):
        # 重新建立数据库连接
        await reconnect_database(new_config['database'])

config_center.watch_config('order-service', on_config_changed)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

# ========== 两阶段提交 (2PC) ==========

class TwoPhaseCommit:
    """两阶段提交协调者"""

    def __init__(self):
        self.participants = []

    def register_participant(self, participant):
        """注册参与者"""
        self.participants.append(participant)

    async def execute(self, transaction_data):
        """执行分布式事务"""
        transaction_id = generate_transaction_id()

        # 阶段1: 准备阶段
        prepared = []
        for participant in self.participants:
            try:
                result = await participant.prepare(transaction_id, transaction_data)
                if result == 'PREPARED':
                    prepared.append(participant)
                else:
                    # 任何参与者拒绝，回滚所有
                    await self._rollback_all(transaction_id, prepared)
                    return False
            except Exception as e:
                await self._rollback_all(transaction_id, prepared)
                raise e

        # 阶段2: 提交阶段
        committed = []
        for participant in prepared:
            try:
                await participant.commit(transaction_id)
                committed.append(participant)
            except Exception as e:
                # 提交失败，需要人工介入
                await self._rollback_all(transaction_id, committed)
                raise Exception(f"Commit failed: {e}")

        return True

    async def _rollback_all(self, transaction_id, participants):
        """回滚所有参与者"""
        for participant in participants:
            try:
                await participant.rollback(transaction_id)
            except Exception as e:
                logging.error(f"Rollback failed: {e}")

# ========== Saga模式 ==========
# 长事务的替代方案

class SagaOrchestrator:
    """Saga编排器"""

    def __init__(self):
        self.steps = []
        self.compensations = []

    def add_step(self, action, compensation):
        """添加步骤"""
        self.steps.append(action)
        self.compensations.append(compensation)

    async def execute(self, initial_data):
        """执行Saga"""
        context = initial_data

        executed_steps = []

        # 执行每个步骤
        for i, step in enumerate(self.steps):
            try:
                context = await step(context)
                executed_steps.append(i)
            except Exception as e:
                # 失败，执行补偿
                await self._compensate(executed_steps, context)
                raise e

        return context

    async def _compensate(self, executed_steps, context):
        """执行补偿事务"""
        # 逆序执行补偿
        for i in reversed(executed_steps):
            try:
                await self.compensations[i](context)
            except Exception as e:
                logging.error(f"Compensation failed: {e}")

# 订单Saga示例
class OrderSaga:
    """订单处理Saga"""

    def __init__(self):
        self.saga = SagaOrchestrator()
        self._setup_steps()

    def _setup_steps(self):
        """设置Saga步骤"""

        # 步骤1: 创建订单
        async def create_order(context):
            order = await order_repository.create(context['order_data'])
            context['order'] = order
            return context

        async def cancel_order(context):
            await order_repository.update_status(
                context['order'].id,
                'CANCELLED'
            )

        # 步骤2: 扣减库存
        async def deduct_inventory(context):
            for item in context['order'].items:
                await inventory_service.deduct_stock(
                    item.product_id,
                    item.quantity
                )
            return context

        async def restore_inventory(context):
            for item in context['order'].items:
                await inventory_service.restore_stock(
                    item.product_id,
                    item.quantity
                )

        # 步骤3: 处理支付
        async def process_payment(context):
            payment = await payment_service.charge(
                context['order'].user_id,
                context['order'].total_amount
            )
            context['payment'] = payment
            return context

        async def refund_payment(context):
            await payment_service.refund(
                context['payment'].transaction_id
            )

        # 步骤4: 发送通知
        async def send_notification(context):
            await notification_service.send(
                context['order'].user_email,
                'Order Created',
                f'Your order {context["order"].id} has been created'
            )
            return context

        async def cancel_notification(context):
            # 通知可能不需要补偿
            pass

        # 添加步骤和补偿
        self.saga.add_step(create_order, cancel_order)
        self.saga.add_step(deduct_inventory, restore_inventory)
        self.saga.add_step(process_payment, refund_payment)
        self.saga.add_step(send_notification, cancel_notification)

    async def execute(self, order_data):
        """执行订单Saga"""
        return await self.saga.execute({'order_data': order_data})

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

# ========== 事件溯源 ==========
# 通过事件流重建状态

class EventStore:
    """事件存储"""

    def __init__(self):
        self.events = []

    async def append_event(self, aggregate_id: str, event: dict):
        """追加事件"""
        event['aggregate_id'] = aggregate_id
        event['timestamp'] = datetime.now().isoformat()
        event['version'] = len(self.events) + 1
        self.events.append(event)

    async def get_events(self, aggregate_id: str) -> List[dict]:
        """获取聚合的所有事件"""
        return [
            e for e in self.events
            if e['aggregate_id'] == aggregate_id
        ]

class OrderAggregate:
    """订单聚合 - 通过事件重建状态"""

    def __init__(self, event_store: EventStore):
        self.event_store = event_store
        self.state = None

    async def rebuild(self, order_id: str):
        """从事件流重建状态"""
        events = await self.event_store.get_events(order_id)

        state = None
        for event in events:
            state = self._apply_event(state, event)

        self.state = state
        return state

    def _apply_event(self, state, event):
        """应用事件到状态"""
        event_type = event['type']

        if event_type == 'OrderCreated':
            return {
                'id': event['order_id'],
                'user_id': event['user_id'],
                'items': event['items'],
                'status': 'CREATED'
            }

        elif event_type == 'PaymentCompleted':
            state['status'] = 'PAID'
            state['payment_id'] = event['payment_id']
            return state

        elif event_type == 'OrderShipped':
            state['status'] = 'SHIPPED'
            state['shipping_id'] = event['shipping_id']
            return state

        elif event_type == 'OrderCancelled':
            state['status'] = 'CANCELLED'
            return state

        return state

    async def create_order(self, user_id, items):
        """创建订单"""
        event = {
            'type': 'OrderCreated',
            'order_id': generate_id(),
            'user_id': user_id,
            'items': items
        }

        await self.event_store.append_event(event['order_id'], event)
        return await self.rebuild(event['order_id'])

# ========== CQRS ==========
# 命令查询职责分离

class CommandBus:
    """命令总线"""

    def __init__(self):
        self.handlers = {}

    def register(self, command_type: str, handler):
        """注册命令处理器"""
        self.handlers[command_type] = handler

    async def execute(self, command: dict):
        """执行命令"""
        command_type = command['type']

        if command_type not in self.handlers:
            raise ValueError(f"Unknown command: {command_type}")

        return await self.handlers[command_type](command)

class QueryBus:
    """查询总线"""

    def __init__(self):
        self.handlers = {}

    def register(self, query_type: str, handler):
        """注册查询处理器"""
        self.handlers[query_type] = handler

    async def execute(self, query: dict):
        """执行查询"""
        query_type = query['type']

        if query_type not in self.handlers:
            raise ValueError(f"Unknown query: {query_type}")

        return await self.handlers[query_type](query)

# CQRS示例
class OrderService:
    """订单服务 - CQRS"""

    def __init__(self):
        self.command_bus = CommandBus()
        self.query_bus = QueryBus()
        self.event_store = EventStore()
        self.read_db = {}  # 读模型

        self._register_handlers()

    def _register_handlers(self):
        """注册处理器"""

        # 命令处理器
        self.command_bus.register('CreateOrder', self._handle_create_order)
        self.command_bus.register('CancelOrder', self._handle_cancel_order)

        # 查询处理器
        self.query_bus.register('GetOrder', self._handle_get_order)
        self.query_bus.register('ListOrders', self._handle_list_orders)

    async def _handle_create_order(self, command):
        """处理创建订单命令"""
        event = {
            'type': 'OrderCreated',
            'order_id': command['order_id'],
            'user_id': command['user_id'],
            'items': command['items']
        }

        await self.event_store.append_event(command['order_id'], event)

        # 更新读模型
        self._update_read_model(event)

        return event['order_id']

    async def _handle_cancel_order(self, command):
        """处理取消订单命令"""
        event = {
            'type': 'OrderCancelled',
            'order_id': command['order_id']
        }

        await self.event_store.append_event(command['order_id'], event)

        # 更新读模型
        self._update_read_model(event)

    async def _handle_get_order(self, query):
        """处理获取订单查询"""
        return self.read_db.get(query['order_id'])

    async def _handle_list_orders(self, query):
        """处理订单列表查询"""
        user_id = query.get('user_id')

        orders = [
            order for order in self.read_db.values()
            if not user_id or order['user_id'] == user_id
        ]

        return orders

    def _update_read_model(self, event):
        """更新读模型"""
        order_id = event['order_id']

        if event['type'] == 'OrderCreated':
            self.read_db[order_id] = {
                'id': order_id,
                'user_id': event['user_id'],
                'items': event['items'],
                'status': 'CREATED'
            }

        elif event['type'] == 'OrderCancelled':
            if order_id in self.read_db:
                self.read_db[order_id]['status'] = 'CANCELLED'

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108

import asyncio
from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = 'CLOSED'      # 正常状态
    OPEN = 'OPEN'          # 熔断状态
    HALF_OPEN = 'HALF_OPEN'  # 半开状态

class CircuitBreaker:
    """熔断器"""

    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: int = 60,
        half_open_attempts: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.half_open_attempts = half_open_attempts

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None

    async def call(self, func, *args, **kwargs):
        """通过熔断器调用函数"""

        if self.state == CircuitState.OPEN:
            # 熔断状态，检查是否可以进入半开
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise CircuitBreakerOpenException(
                    f"Circuit breaker is OPEN. Try again later."
                )

        try:
            result = await func(*args, **kwargs)

            # 成功，重置计数
            self._on_success()

            return result

        except Exception as e:
            # 失败，增加计数
            self._on_failure()
            raise e

    def _should_attempt_reset(self) -> bool:
        """检查是否应该尝试重置"""
        if self.last_failure_time is None:
            return False

        elapsed = (datetime.now() - self.last_failure_time).seconds
        return elapsed >= self.timeout

    def _on_success(self):
        """处理成功"""
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1

            # 半开状态下连续成功，恢复关闭状态
            if self.success_count >= self.half_open_attempts:
                self.state = CircuitState.CLOSED
                self.failure_count = 0

        elif self.state == CircuitState.CLOSED:
            self.failure_count = 0

    def _on_failure(self):
        """处理失败"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()

        # 达到阈值，打开熔断器
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# 使用示例
async def call_external_service(url):
    """调用外部服务"""
    response = await aiohttp.get(url)
    return await response.json()

# 创建熔断器
circuit_breaker = CircuitBreaker(
    failure_threshold=5,
    timeout=60,
    half_open_attempts=3
)

# 通过熔断器调用
try:
    result = await circuit_breaker.call(
        call_external_service,
        'http://external-service/api/data'
    )
except CircuitBreakerOpenException:
    # 熔断器打开，使用降级逻辑
    result = get_cached_data()
except Exception as e:
    # 其他错误处理
    logger.error(f"Service call failed: {e}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103

import asyncio
from functools import wraps
from typing import Callable, Type

class RetryConfig:
    """重试配置"""

    def __init__(
        self,
        max_attempts: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 10.0,
        exponential_base: float = 2,
        jitter: bool = True,
        retry_exceptions: list = None
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_base = exponential_base
        self.jitter = jitter
        self.retry_exceptions = retry_exceptions or [Exception]

def retry(config: RetryConfig = None):
    """重试装饰器"""

    if config is None:
        config = RetryConfig()

    def decorator(func: Callable):

        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(1, config.max_attempts + 1):
                try:
                    return await func(*args, **kwargs)

                except tuple(config.retry_exceptions) as e:
                    last_exception = e

                    if attempt < config.max_attempts:
                        # 计算延迟时间
                        delay = min(
                            config.base_delay * (config.exponential_base ** (attempt - 1)),
                            config.max_delay
                        )

                        # 添加抖动
                        if config.jitter:
                            delay = delay * (0.5 + random.random() * 0.5)

                        logger.warning(
                            f"Attempt {attempt} failed: {e}. "
                            f"Retrying in {delay:.2f}s..."
                        )

                        await asyncio.sleep(delay)

            # 所有尝试都失败
            raise last_exception

        return wrapper

    return decorator

# 超时装饰器
def timeout(seconds: float):
    """超时装饰器"""

    def decorator(func: Callable):

        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(
                    func(*args, **kwargs),
                    timeout=seconds
                )
            except asyncio.TimeoutError:
                raise TimeoutException(
                    f"Function {func.__name__} timed out after {seconds}s"
                )

        return wrapper

    return decorator

# 使用示例
@retry(RetryConfig(
    max_attempts=3,
    base_delay=1.0,
    exponential_base=2,
    retry_exceptions=[ConnectionError, TimeoutError]
))
@timeout(seconds=5)
async def call_external_api(url):
    """调用外部API，带重试和超时"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

import time
from collections import deque
from typing import Callable, Any

class RateLimiter:
    """速率限制器"""

    def __init__(self, rate: int, per: float):
        """
        rate: 允许的请求数
        per: 时间窗口（秒）
        """
        self.rate = rate
        self.per = per
        self.allowance = rate
        self.last_check = time.time()

    def acquire(self, tokens: int = 1) -> bool:
        """获取令牌"""
        current = time.time()
        elapsed = current - self.last_check

        # 补充令牌
        self.allowance += elapsed * (self.rate / self.per)

        if self.allowance > self.rate:
            self.allowance = self.rate

        self.last_check = current

        # 检查是否有足够的令牌
        if self.allowance < tokens:
            return False

        self.allowance -= tokens
        return True

class TokenBucket:
    """令牌桶算法"""

    def __init__(self, capacity: int, refill_rate: float):
        """
        capacity: 桶容量
        refill_rate: 填充速率（每秒）
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.time()

    def consume(self, tokens: int = 1) -> bool:
        """消费令牌"""
        self._refill()

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True

        return False

    def _refill(self):
        """补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill

        refill_amount = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + refill_amount)

        self.last_refill = now

class SlidingWindow:
    """滑动窗口限流"""

    def __init__(self, limit: int, window: float):
        """
        limit: 窗口内最大请求数
        window: 时间窗口（秒）
        """
        self.limit = limit
        self.window = window
        self.requests = deque()

    def is_allowed(self) -> bool:
        """检查是否允许请求"""
        now = time.time()

        # 移除窗口外的请求
        while self.requests and self.requests[0] < now - self.window:
            self.requests.popleft()

        # 检查是否超过限制
        if len(self.requests) >= self.limit:
            return False

        self.requests.append(now)
        return True

# 降级装饰器
class FallbackExecutor:
    """降级执行器"""

    def __init__(self):
        self.fallbacks = {}

    def register_fallback(self, func_name: str, fallback: Callable):
        """注册降级函数"""
        self.fallbacks[func_name] = fallback

    async def execute_with_fallback(
        self,
        func: Callable,
        *args,
        fallback_result: Any = None,
        **kwargs
    ):
        """执行函数，失败时降级"""
        try:
            return await func(*args, **kwargs)

        except Exception as e:
            func_name = func.__name__

            # 查找注册的降级函数
            if func_name in self.fallbacks:
                logger.warning(f"Function {func_name} failed, using fallback")
                return await self.fallbacks[func_name](*args, **kwargs)

            # 使用默认降级结果
            if fallback_result is not None:
                logger.warning(f"Function {func_name} failed, using fallback result")
                return fallback_result

            # 没有降级方案，抛出异常
            raise e

# 使用示例
# 创建限流器
rate_limiter = RateLimiter(rate=100, per=1)  # 100请求/秒
token_bucket = TokenBucket(capacity=10, refill_rate=1)  # 10令牌容量，每秒补充1个
sliding_window = SlidingWindow(limit=100, window=60)  # 60秒内最多100请求

# 创建降级执行器
fallback_executor = FallbackExecutor()

async def get_user_data(user_id):
    """获取用户数据"""
    # 检查限流
    if not rate_limiter.acquire():
        raise RateLimitException("Too many requests")

    # 调用服务
    return await user_service.get_user(user_id)

# 注册降级函数
async def get_user_data_fallback(user_id):
    """降级：返回缓存的用户数据"""
    return await cache.get(f"user:{user_id}")

fallback_executor.register_fallback('get_user_data', get_user_data_fallback)

# 使用
try:
    result = await fallback_executor.execute_with_fallback(
        get_user_data,
        user_id='123'
    )
except Exception as e:
    logger.error(f"All attempts failed: {e}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

import structlog
from typing import Any

class LogContext:
    """日志上下文"""

    def __init__(self):
        self.context = {}

    def set(self, key: str, value: Any):
        """设置上下文"""
        self.context[key] = value

    def get(self, key: str, default=None):
        """获取上下文"""
        return self.context.get(key, default)

    def clear(self):
        """清空上下文"""
        self.context.clear()

# 全局日志上下文
log_context = LogContext()

# 配置structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        # 添加上下文
        lambda logger, method_name, event_dict: {
            **event_dict,
            **log_context.context
        },
        # 格式化输出
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

class ServiceLogger:
    """服务日志记录器"""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.logger = structlog.get_logger()

    def log_request(self, request_id: str, method: str, path: str, **kwargs):
        """记录请求"""
        self.logger.info(
            "incoming_request",
            request_id=request_id,
            service=self.service_name,
            method=method,
            path=path,
            **kwargs
        )

    def log_response(
        self,
        request_id: str,
        status_code: int,
        duration_ms: float,
        **kwargs
    ):
        """记录响应"""
        self.logger.info(
            "outgoing_response",
            request_id=request_id,
            service=self.service_name,
            status_code=status_code,
            duration_ms=duration_ms,
            **kwargs
        )

    def log_error(self, error: Exception, **kwargs):
        """记录错误"""
        self.logger.error(
            "error_occurred",
            service=self.service_name,
            error_type=type(error).__name__,
            error_message=str(error),
            **kwargs
        )

    def log_service_call(
        self,
        service_name: str,
        method: str,
        duration_ms: float,
        success: bool,
        **kwargs
    ):
        """记录服务调用"""
        self.logger.info(
            "service_call",
            caller=self.service_name,
            service=service_name,
            method=method,
            duration_ms=duration_ms,
            success=success,
            **kwargs
        )

# 中间件示例
class LoggingMiddleware:
    """日志中间件"""

    def __init__(self, logger: ServiceLogger):
        self.logger = logger

    async def process_request(self, request, call_next):
        """处理请求"""
        request_id = generate_request_id()
        start_time = time.time()

        # 设置日志上下文
        log_context.set('request_id', request_id)
        log_context.set('user_id', request.user_id)

        # 记录请求
        self.logger.log_request(
            request_id=request_id,
            method=request.method,
            path=request.path
        )

        try:
            # 处理请求
            response = await call_next(request)

            # 记录响应
            duration_ms = (time.time() - start_time) * 1000
            self.logger.log_response(
                request_id=request_id,
                status_code=response.status_code,
                duration_ms=duration_ms
            )

            return response

        except Exception as e:
            duration_ms = (time.time() - start_time) * 1000
            self.logger.log_error(
                error=e,
                request_id=request_id,
                duration_ms=duration_ms
            )
            raise

        finally:
            log_context.clear()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 配置Tracer
trace.set_tracer_provider(TracerProvider())
tracer_provider = trace.get_tracer_provider()

# 配置Jaeger导出器
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

tracer_provider.add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

class TracingClient:
    """带追踪的客户端"""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.tracer = trace.get_tracer(__name__)

    async def call_service(
        self,
        service_name: str,
        method: str,
        **kwargs
    ):
        """调用服务并追踪"""

        with self.tracer.start_as_current_span(
            f"{service_name}.{method}",
            kind=trace.SpanKind.CLIENT
        ) as span:
            # 添加属性
            span.set_attribute("service", self.service_name)
            span.set_attribute("target_service", service_name)
            span.set_attribute("method", method)

            try:
                # 注入追踪上下文
                headers = {}
                trace.inject(headers)

                # 调用服务
                result = await self._make_request(
                    service_name,
                    method,
                    headers=headers,
                    **kwargs
                )

                span.set_attribute("success", True)
                return result

            except Exception as e:
                span.record_exception(e)
                span.set_attribute("success", False)
                raise

    async def _make_request(self, service_name, method, headers, **kwargs):
        """实际请求逻辑"""
        # 实现服务调用
        pass

# 使用示例
client = TracingClient("order-service")

async def create_order(user_id, items):
    """创建订单 - 带追踪"""

    with client.tracer.start_as_current_span("create_order") as span:
        span.set_attribute("user_id", user_id)
        span.set_attribute("item_count", len(items))

        # 调用库存服务
        inventory_result = await client.call_service(
            "inventory-service",
            "check_stock",
            items=items
        )

        # 调用支付服务
        payment_result = await client.call_service(
            "payment-service",
            "process_payment",
            user_id=user_id,
            amount=calculate_amount(items)
        )

        return {
            "inventory": inventory_result,
            "payment": payment_result
        }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
request_count = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

request_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint']
)

active_connections = Gauge(
    'active_connections',
    'Number of active connections'
)

business_metric = Counter(
    'business_operations_total',
    'Total business operations',
    ['operation', 'status']
)

class MetricsMiddleware:
    """指标收集中间件"""

    def __init__(self):
        self.active_connections = active_connections

    async def process_request(self, request, call_next):
        """处理请求并收集指标"""

        # 增加活跃连接数
        self.active_connections.inc()

        start_time = time.time()

        try:
            response = await call_next(request)

            # 记录请求计数
            request_count.labels(
                method=request.method,
                endpoint=request.path,
                status=response.status_code
            ).inc()

            # 记录请求耗时
            duration = time.time() - start_time
            request_duration.labels(
                method=request.method,
                endpoint=request.path
            ).observe(duration)

            return response

        finally:
            # 减少活跃连接数
            self.active_connections.dec()

class BusinessMetrics:
    """业务指标收集"""

    @staticmethod
    def record_operation(operation: str, success: bool):
        """记录业务操作"""
        status = "success" if success else "failure"
        business_metric.labels(
            operation=operation,
            status=status
        ).inc()

    @staticmethod
    def record_order_created(order_value: float):
        """记录订单创建"""
        business_metric.labels(
            operation="order_created",
            status="success"
        ).inc()

    @staticmethod
    def record_payment_failed(amount: float, reason: str):
        """记录支付失败"""
        business_metric.labels(
            operation=f"payment_failed_{reason}",
            status="failure"
        ).inc()

# 使用示例
async def create_order_logic(user_id, items):
    """创建订单逻辑"""

    try:
        order = await order_service.create(user_id, items)

        BusinessMetrics.record_operation("order_created", True)

        return order

    except InventoryError as e:
        BusinessMetrics.record_operation("order_created", False)
        BusinessMetrics.record_operation("inventory_check_failed", False)
        raise

    except PaymentError as e:
        BusinessMetrics.record_operation("order_created", False)
        BusinessMetrics.record_payment_failed(
            order.total,
            e.reason
        )
        raise

# 启动指标服务器
start_http_server(8000)