本文目录导读：
我来详细解释Python事件驱动架构的搭建方法,从基础到进阶。
基础事件驱动模型
最简单的回调模式
from typing import Callable, Dict, List, Any
import time
class SimpleEventBus:
def __init__(self):
self.handlers: Dict[str, List[Callable]] = {}
def register(self, event_type: str, handler: Callable):
if event_type not in self.handlers:
self.handlers[event_type] = []
self.handlers[event_type].append(handler)
def emit(self, event_type: str, data: Any = None):
if event_type in self.handlers:
for handler in self.handlers[event_type]:
handler(data)
# 使用示例
def on_user_login(data):
print(f"User login: {data}")
def on_order_placed(data):
print(f"Order placed: {data}")
bus = SimpleEventBus()
bus.register("user.login", on_user_login)
bus.register("order.placed", on_order_placed)
bus.emit("user.login", {"username": "alice", "time": time.time()})
bus.emit("order.placed", {"order_id": "123", "amount": 99.99})
进阶事件驱动架构
使用asyncio的事件循环
import asyncio
from typing import Dict, List, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class Event:
"""事件数据类"""
type: str
data: Any
timestamp: datetime = field(default_factory=datetime.now)
source: str = ""
correlation_id: str = ""
class AsyncEventBus:
"""异步事件总线"""
def __init__(self):
self.handlers: Dict[str, List[Callable]] = {}
self.middlewares: List[Callable] = []
def register(self, event_type: str, handler: Callable):
"""注册事件处理器"""
if event_type not in self.handlers:
self.handlers[event_type] = []
self.handlers[event_type].append(handler)
logger.info(f"Handler registered for event: {event_type}")
def unregister(self, event_type: str, handler: Callable):
"""移除事件处理器"""
if event_type in self.handlers:
self.handlers[event_type].remove(handler)
def add_middleware(self, middleware: Callable):
"""添加中间件"""
self.middlewares.append(middleware)
async def emit(self, event: Event):
"""触发事件"""
logger.info(f"Event emitted: {event.type}")
# 应用中间件
for middleware in self.middlewares:
event = await middleware(event)
if event is None:
return
# 处理事件
if event.type in self.handlers:
tasks = []
for handler in self.handlers[event.type]:
tasks.append(handler(event))
await asyncio.gather(*tasks)
# 使用示例
class UserEventHandler:
async def handle_login(self, event: Event):
await asyncio.sleep(0.1) # 模拟异步操作
logger.info(f"User login handled: {event.data}")
async def handle_logout(self, event: Event):
logger.info(f"User logout handled: {event.data}")
async def main():
bus = AsyncEventBus()
handler = UserEventHandler()
# 注册处理器
bus.register("user.login", handler.handle_login)
bus.register("user.logout", handler.handle_logout)
# 触发事件
await bus.emit(Event(
type="user.login",
data={"user_id": 1, "username": "alice"},
source="auth_service"
))
await bus.emit(Event(
type="user.logout",
data={"user_id": 1},
source="web_session"
))
# 运行
asyncio.run(main())
企业级事件驱动架构
完整的生产级实现
import asyncio
from typing import Dict, List, Any, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
import json
import logging
import uuid
from enum import Enum
from collections import defaultdict
import traceback
from concurrent.futures import ThreadPoolExecutor
import pickle
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("EventDrivenArchitecture")
class EventPriority(Enum):
"""事件优先级"""
CRITICAL = 1
HIGH = 2
NORMAL = 3
LOW = 4
class EventStatus(Enum):
"""事件状态"""
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
RETRYING = "retrying"
@dataclass
class Event:
"""企业级事件对象"""
type: str
data: Any
priority: EventPriority = EventPriority.NORMAL
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: datetime = field(default_factory=datetime.now)
source: str = ""
correlation_id: str = ""
status: EventStatus = EventStatus.PENDING
retry_count: int = 0
max_retries: int = 3
def to_dict(self) -> dict:
return {
"event_id": self.event_id,
"type": self.type,
"data": self.data,
"priority": self.priority.value,
"timestamp": self.timestamp.isoformat(),
"source": self.source,
"correlation_id": self.correlation_id,
"status": self.status.value,
"retry_count": self.retry_count
}
class EventHandler:
"""事件处理器基类"""
async def handle(self, event: Event) -> bool:
"""处理事件，返回是否成功"""
raise NotImplementedError
@property
def event_types(self) -> Set[str]:
"""返回处理器支持的事件类型"""
raise NotImplementedError
@property
def name(self) -> str:
return self.__class__.__name__
class EventMiddleware:
"""事件中间件基类"""
async def process(self, event: Event) -> Optional[Event]:
"""处理事件，返回None表示中断"""
return event
class EventBus:
"""企业级事件总线"""
def __init__(self, max_workers: int = 10):
self.handlers: Dict[str, List[EventHandler]] = defaultdict(list)
self.middlewares: List[EventMiddleware] = []
self.priority_queues: Dict[EventPriority, asyncio.Queue] = {
priority: asyncio.Queue() for priority in EventPriority
}
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.stats = {
"total_events": 0,
"successful_events": 0,
"failed_events": 0,
"retried_events": 0
}
def register_handler(self, handler: EventHandler):
"""注册事件处理器"""
for event_type in handler.event_types:
self.handlers[event_type].append(handler)
logger.info(f"Handler '{handler.name}' registered for event: {event_type}")
def unregister_handler(self, handler: EventHandler):
"""移除事件处理器"""
for event_type in list(self.handlers.keys()):
if handler in self.handlers[event_type]:
self.handlers[event_type].remove(handler)
def add_middleware(self, middleware: EventMiddleware):
"""添加中间件"""
self.middlewares.append(middleware)
logger.info(f"Middleware added: {middleware.__class__.__name__}")
async def emit(self, event: Event):
"""发布事件到队列"""
self.stats["total_events"] += 1
# 应用中间件
for middleware in self.middlewares:
try:
event = await middleware.process(event)
if event is None:
logger.info(f"Event {event.event_id} filtered by middleware")
return
except Exception as e:
logger.error(f"Middleware error: {e}")
return
# 放入优先级队列
await self.priority_queues[event.priority].put(event)
async def process_events(self):
"""处理事件的主循环"""
while True:
# 按优先级处理事件
for priority in EventPriority:
queue = self.priority_queues[priority]
try:
# 非阻塞获取事件
event = queue.get_nowait()
except asyncio.QueueEmpty:
continue
# 异步处理事件
asyncio.create_task(self._process_event(event))
await asyncio.sleep(0.01) # 避免CPU过载
async def _process_event(self, event: Event):
"""处理单个事件"""
event.status = EventStatus.PROCESSING
handlers = self.handlers.get(event.type, [])
if not handlers:
logger.warning(f"No handlers for event type: {event.type}")
event.status = EventStatus.COMPLETED
return
success = True
for handler in handlers:
try:
result = await handler.handle(event)
if not result:
success = False
except Exception as e:
logger.error(f"Handler '{handler.name}' failed: {e}\n{traceback.format_exc()}")
success = False
if success:
event.status = EventStatus.COMPLETED
self.stats["successful_events"] += 1
logger.info(f"Event {event.event_id} processed successfully")
else:
await self._handle_failed_event(event)
async def _handle_failed_event(self, event: Event):
"""处理失败事件"""
if event.retry_count < event.max_retries:
event.retry_count += 1
event.status = EventStatus.RETRYING
self.stats["retried_events"] += 1
# 指数退避重试
wait_time = 2 ** event.retry_count
logger.info(f"Retrying event {event.event_id} in {wait_time}s")
await asyncio.sleep(wait_time)
# 重新放入队列
await self.emit(event)
else:
event.status = EventStatus.FAILED
self.stats["failed_events"] += 1
logger.error(f"Event {event.event_id} failed after {event.retry_count} retries")
def get_stats(self) -> dict:
"""获取统计信息"""
return {
**self.stats,
"queue_sizes": {
priority.name: queue.qsize()
for priority, queue in self.priority_queues.items()
}
}
class LoggingMiddleware(EventMiddleware):
"""日志中间件"""
async def process(self, event: Event) -> Optional[Event]:
logger.info(f"Event received: {event.type} [{event.priority.name}]")
return event
class ValidationMiddleware(EventMiddleware):
"""验证中间件"""
async def process(self, event: Event) -> Optional[Event]:
if not event.type:
logger.warning("Event without type filtered")
return None
if event.priority not in EventPriority:
event.priority = EventPriority.NORMAL
return event
# 具体事件处理器示例
class UserLoginHandler(EventHandler):
async def handle(self, event: Event) -> bool:
logger.info(f"Processing user login: {event.data.get('username')}")
# 模拟处理
await asyncio.sleep(0.1)
return True
@property
def event_types(self) -> Set[str]:
return {"user.login"}
@property
def name(self) -> str:
return "UserLoginHandler"
class OrderHandler(EventHandler):
async def handle(self, event: Event) -> bool:
logger.info(f"Processing order: {event.data.get('order_id')}")
# 模拟失败的情况
if event.retry_count > 0 and event.retry_count < 2:
return False
await asyncio.sleep(0.2)
return True
@property
def event_types(self) -> Set[str]:
return {"order.created", "order.updated"}
@property
def name(self) -> str:
return "OrderHandler"
async def main():
# 初始化事件总线
bus = EventBus(max_workers=5)
# 添加中间件
bus.add_middleware(LoggingMiddleware())
bus.add_middleware(ValidationMiddleware())
# 注册事件处理器
bus.register_handler(UserLoginHandler())
bus.register_handler(OrderHandler())
# 启动事件处理循环
asyncio.create_task(bus.process_events())
# 发送测试事件
events = [
Event(
type="user.login",
data={"username": "alice", "ip": "192.168.1.1"},
priority=EventPriority.HIGH,
source="web_interface"
),
Event(
type="order.created",
data={"order_id": "ORD-001", "amount": 299.99},
priority=EventPriority.NORMAL,
source="api_gateway"
),
Event(
type="order.updated",
data={"order_id": "ORD-001", "status": "paid"},
priority=EventPriority.LOW,
source="payment_service"
)
]
# 发布事件
for event in events:
await bus.emit(event)
# 等待事件处理完成
await asyncio.sleep(2)
# 打印统计信息
print("\n=== Event Bus Statistics ===")
stats = bus.get_stats()
for key, value in stats.items():
print(f"{key}: {value}")
if __name__ == "__main__":
asyncio.run(main())
高级特性：事件溯源（Event Sourcing）
from typing import List, Dict, Any
from abc import ABC, abstractmethod
import json
from datetime import datetime
class EventStore(ABC):
@abstractmethod
async def save_events(self, aggregate_id: str, events: List[Event], expected_version: int):
pass
@abstractmethod
async def get_events(self, aggregate_id: str) -> List[Event]:
pass
class InMemoryEventStore(EventStore):
def __init__(self):
self.events: Dict[str, List[Event]] = defaultdict(list)
self.versions: Dict[str, int] = {}
async def save_events(self, aggregate_id: str, events: List[Event], expected_version: int):
current_version = self.versions.get(aggregate_id, 0)
if current_version != expected_version:
raise ValueError("Concurrency conflict")
self.events[aggregate_id].extend(events)
self.versions[aggregate_id] = expected_version + len(events)
async def get_events(self, aggregate_id: str) -> List[Event]:
return self.events.get(aggregate_id, [])
class Aggregate(ABC):
def __init__(self, id: str):
self.id = id
self.version = 0
self.changes: List[Event] = []
@abstractmethod
def apply_event(self, event: Event):
pass
def load_from_events(self, events: List[Event]):
for event in events:
self.apply_event(event)
self.version += 1
def add_event(self, event: Event):
self.changes.append(event)
self.apply_event(event)
self.version += 1
class UserAggregate(Aggregate):
def __init__(self, id: str):
super().__init__(id)
self.username = ""
self.email = ""
self.is_active = False
def create_user(self, username: str, email: str):
self.add_event(Event(
type="user.created",
data={"username": username, "email": email},
source="UserAggregate"
))
def deactivate_user(self):
self.add_event(Event(
type="user.deactivated",
data={"user_id": self.id},
source="UserAggregate"
))
def apply_event(self, event: Event):
if event.type == "user.created":
self.username = event.data["username"]
self.email = event.data["email"]
self.is_active = True
elif event.type == "user.deactivated":
self.is_active = False
使用第三方库简化开发
使用 RabbitMQ 实现分布式事件系统
import aio_pika
import asyncio
import json
from typing import Callable
class RabbitMQEventBus:
def __init__(self, url: str = "amqp://guest:guest@localhost/"):
self.url = url
self.connection = None
self.channel = None
self.exchange = None
async def connect(self):
self.connection = await aio_pika.connect_robust(self.url)
self.channel = await self.connection.channel()
self.exchange = await self.channel.declare_exchange(
"events", aio_pika.ExchangeType.TOPIC
)
logger.info("Connected to RabbitMQ")
async def publish(self, routing_key: str, message: dict):
await self.exchange.publish(
aio_pika.Message(body=json.dumps(message).encode()),
routing_key=routing_key
)
async def subscribe(self, routing_key: str, callback: Callable):
queue = await self.channel.declare_queue(exclusive=True)
await queue.bind(self.exchange, routing_key=routing_key)
async with queue.iterator() as queue_iter:
async for message in queue_iter:
async with message.process():
data = json.loads(message.body.decode())
await callback(data)
async def close(self):
if self.connection:
await self.connection.close()
最佳实践建议
-
选择合适的实现方式：
- 小型应用：使用简单的回调模式
- 中型应用：使用 asyncio 实现
- 大型分布式系统：使用消息队列（RabbitMQ, Kafka）
-
错误处理策略：
- 实现重试机制和指数退避
- 记录失败事件到死信队列
- 监控事件处理延迟
-
性能优化：
- 使用异步架构提高吞吐量
- 实现事件的批量处理
- 合理设置优先级队列
-
可观测性：
- 添加事件追踪ID
- 实现完整的事件审计日志
- 监控关键指标（事件处理时间、失败率等）
这个架构可以根据你的具体需求进行调整和扩展,从小型应用到企业级系统都能适用。
