Python应用事件驱动架构怎么落地

Python应用事件驱动架构落地实践

核心概念与架构选择

事件驱动架构的核心是事件生产者、事件总线、事件消费者的三角关系，在Python中,可以从轻量级到企业级选择不同的实现路径：

方案对比

实践方案一：单进程事件中心（推荐初创项目）

核心实现（基于blinker）

# event_center.py
from blinker import signal
import json
# 1. 定义命名空间事件
order_created = signal('order-created')
order_paid = signal('order-paid')
# 2. 装饰器式注册监听器
@order_created.connect
def handle_order_created(sender, **kwargs):
    order = kwargs['order']
    print(f"[Event] 订单创建: {order['id']}")
    # 触发后续事件
    check_inventory(order)
@order_created.connect
def send_notification(sender, **kwargs):
    order = kwargs['order']
    print(f"[Event] 发送通知: 订单{order['id']}已创建")
# 3. 生产者发布事件
def create_order(order_data):
    # 业务逻辑...
    order = {'id': '20230101', 'items': ['book', 'pen'], 'total': 100}
    order_created.send(order=order)  # 同步触发
# 4. 异步事件处理（可选）
from threading import Thread
def async_listener(sender, **kwargs):
    Thread(target=process_async, args=(kwargs,)).start()
order_created.connect(async_listener, weak=False)

异步版本（基于asyncio）

import asyncio
from collections import defaultdict
class AsyncEventBus:
    def __init__(self):
        self._handlers = defaultdict(list)
    def register(self, event, handler):
        self._handlers[event].append(handler)
    def unregister(self, event, handler):
        self._handlers[event].remove(handler)
    async def emit(self, event, **data):
        tasks = [handler(**data) for handler in self._handlers[event]]
        await asyncio.gather(*tasks)
    async def emit_sequential(self, event, **data):
        for handler in self._handlers[event]:
            await handler(**data)
# 使用示例
bus = AsyncEventBus()
@bus.register('user.login')
async def update_login_count(user_id):
    await asyncio.sleep(0.1)
    print(f"更新用户{user_id}登录次数")
@bus.register('user.login')
async def update_last_login_time(user_id):
    await asyncio.sleep(0.2)
    print(f"更新用户{user_id}最后登录时间")
async def main():
    await bus.emit('user.login', user_id=123)
asyncio.run(main())

实践方案二：分布式事件系统（推荐中大型系统）

基于Redis Pub/Sub（适合中小规模）

# redis_event_bus.py
import json
import redis
import threading
class RedisEventBus:
    def __init__(self, redis_url='redis://localhost:6379/0'):
        self.client = redis.from_url(redis_url)
        self.pubsub = self.client.pubsub()
        self._handlers = {}
        self._listener_thread = None
    def register(self, channel, handler):
        if channel not in self._handlers:
            self._handlers[channel] = []
            self.pubsub.subscribe(**{channel: self._message_handler})
        self._handlers[channel].append(handler)
    def _message_handler(self, message):
        channel = message['channel']
        data = json.loads(message['data'])
        for handler in self._handlers.get(channel, []):
            handler(data)
    def publish(self, channel, data):
        self.client.publish(channel, json.dumps(data))
    def start_listener(self):
        self._listener_thread = threading.Thread(target=self.pubsub.run_in_thread, daemon=True)
        self._listener_thread.start()
    def stop(self):
        if self._listener_thread:
            self.pubsub.close()
# 使用
bus = RedisEventBus()
bus.register('order.created', lambda data: print(f"订单创建事件: {data}"))
bus.start_listener()
bus.publish('order.created', {'order_id': '123', 'total': 100})

基于RabbitMQ（生产级）

# rabbitmq_event_bus.py
import pika
import json
import threading
class RabbitMQEventBus:
    EXCHANGE_NAME = 'event_bus'
    def __init__(self, connection_params):
        self.params = connection_params
        self.connection = None
        self.channel = None
        self.consumers = {}
    def connect(self):
        self.connection = pika.BlockingConnection(self.params)
        self.channel = self.connection.channel()
        self.channel.exchange_declare(exchange=self.EXCHANGE_NAME, 
                                      exchange_type='topic')
    def publish(self, event_type, data, routing_key=None):
        routing_key = routing_key or event_type
        self.channel.basic_publish(
            exchange=self.EXCHANGE_NAME,
            routing_key=routing_key,
            body=json.dumps({'event_type': event_type, 'data': data})
        )
    def subscribe(self, binding_keys, callback):
        result = self.channel.queue_declare('', exclusive=True)
        queue_name = result.method.queue
        for binding_key in binding_keys:
            self.channel.queue_bind(
                exchange=self.EXCHANGE_NAME,
                queue=queue_name,
                routing_key=binding_key
            )
        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=lambda ch, method, properties, body: 
                callback(json.loads(body.decode()))
        )
    def start_consuming(self):
        self.channel.start_consuming()
# 生产者（订单服务）
bus = RabbitMQEventBus(pika.ConnectionParameters('localhost'))
bus.connect()
bus.publish('order.created', {'order_id': 1})
# 消费者（库存服务）
def on_order_created(message):
    order = message['data']
    print(f"库存服务: 订单{order['order_id']}触发库存检查")
bus2 = RabbitMQEventBus(pika.ConnectionParameters('localhost'))
bus2.connect()
bus2.subscribe(['order.*'], on_order_created)
threading.Thread(target=bus2.start_consuming, daemon=True).start()

实践方案三：事件溯源（适合需要审计和重放的系统）

# event_sourcing_example.py
from eventsourcing.domain import Aggregate, event
from eventsourcing.application import SQLAlchemyApplication
class Order(Aggregate):
    @event('OrderCreated')
    def __init__(self, customer_id):
        self.customer_id = customer_id
        self.items = []
        self.status = 'pending'
    @event('ItemAdded')
    def add_item(self, product_id, quantity):
        self.items.append({'product_id': product_id, 'quantity': quantity})
    @event('OrderPaid')
def pay(self, payment_id):
        self.status = 'paid'
        self.payment_id = payment_id
    def __repr__(self):
        return f"Order(id={self.id}, status={self.status})"
# 使用示例
app = SQLAlchemyApplication()
order = Order(customer_id='user1')
order.add_item('product_1', 2)
order.add_item('product_2', 1)
order.pay('payment_123')
app.save(order)  # 保存所有事件到数据库
# 重建聚合状态
order2 = app.repository.get(order.id)
print(order2.items)  # 输出完整状态

架构设计关键点

事件契约设计

from dataclasses import dataclass
from typing import Optional
@dataclass
class Event:
    event_id: str
    event_type: str
    timestamp: float
    source_service: str
    data: dict
    correlation_id: Optional[str] = None
# 使用Pydantic或Avro进行验证

错误处理策略

# 死信队列模式
class ResilientEventBus:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
        self._dead_letters = []
    def publish_with_retry(self, event):
        for attempt in range(self.max_retries):
            try:
                self._actually_publish(event)
                return
            except Exception as e:
                if attempt == self.max_retries - 1:
                    self._dead_letters.append((event, e))
                else:
                    time.sleep(2 ** attempt)  # 指数退避

幂等性保障

# Redis去重
import redis
from functools import wraps
def idempotent_listener(redis_client, ttl=3600):
    def decorator(func):
        @wraps(func)
        def wrapper(event):
            event_id = event.get('event_id')
            if event_id:
                if redis_client.sismember('processed_events', event_id):
                    return
                redis_client.sadd('processed_events', event_id)
                redis_client.expire('processed_events', ttl)
            return func(event)
        return wrapper
    return decorator

监控与可观测性

# OpenTelemetry集成
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
tracer = trace.get_tracer(__name__)
class ObservableEventBus:
    def publish(self, event):
        with tracer.start_as_current_span(event.event_type) as span:
            span.set_attribute('event_id', event.event_id)
            # 实际发布逻辑

落地步骤建议

1. 初期（1-3个月）：使用进程内blinker或asyncio.Queue，快速实现模块解耦
2. 中期（3-6个月）：引入Redis Pub/Sub支持跨服务的事件通信
3. 成熟期（6个月+）：使用Kafka/RabbitMQ实现持久化、顺序保证、分区等高级特性

常见陷阱

• 事件顺序丢失：使用Kafka分区保证同一订单事件有序
• 循环事件：设置最大递归深度或事件跳跃检测
• 调试困难：使用事件追踪ID（correlation_id）串联所有相关事件
• 性能瓶颈：避免事件总线成为单点故障，使用消息中间件集群

1. 事件不可变：一旦发布不应修改
2. 最少保证：根据业务选择At-most-once, At-least-once, Exactly-once
3. 关注点分离：事件结构不应包含服务内部实现细节
4. 尽早失败：事件验证失败立即报错，不要静默丢弃

通过合理选择技术栈和架构模式，Python完全可以高效实现事件驱动架构，建议从简单方案开始,根据业务增长逐步演进。