当前位置：首页 > news >正文

Python事件驱动架构：从基础到生产实践

news 2026/5/10 9:35:08

Python事件驱动架构：从基础到生产实践

引言

事件驱动架构是一种设计模式，通过事件的产生、传播和处理来实现组件间的解耦。在高并发场景下，事件驱动架构能够提供更好的可扩展性和响应性。

Python提供了多种事件驱动编程的工具和框架。本文将深入探讨事件驱动架构的核心概念，并分享在生产环境中的实践经验。

一、事件驱动架构基础

1.1 核心概念

class Event: """事件基类""" def __init__(self, event_type, data=None): self.event_type = event_type self.data = data self.timestamp = datetime.now() class EventHandler: """事件处理器接口""" def handle(self, event): raise NotImplementedError class EventBus: """事件总线 - 管理事件发布和订阅""" def __init__(self): self.handlers = defaultdict(list) def subscribe(self, event_type, handler): self.handlers[event_type].append(handler) def publish(self, event): for handler in self.handlers.get(event.event_type, []): handler.handle(event)

1.2 同步事件处理

from collections import defaultdict from datetime import datetime class UserCreatedHandler(EventHandler): def handle(self, event): print(f"处理用户创建事件: {event.data}") class NotificationHandler(EventHandler): def handle(self, event): print(f"发送通知: {event.data}") # 使用事件总线 event_bus = EventBus() event_bus.subscribe('user.created', UserCreatedHandler()) event_bus.subscribe('user.created', NotificationHandler()) # 发布事件 event = Event('user.created', {'user_id': 1, 'name': 'John'}) event_bus.publish(event)

二、异步事件驱动

2.1 使用asyncio实现异步事件处理

import asyncio from collections import defaultdict class AsyncEventBus: def __init__(self): self.handlers = defaultdict(list) def subscribe(self, event_type, handler): self.handlers[event_type].append(handler) async def publish(self, event): tasks = [] for handler in self.handlers.get(event.event_type, []): tasks.append(handler.handle(event)) await asyncio.gather(*tasks) class AsyncEmailHandler(EventHandler): async def handle(self, event): await asyncio.sleep(1) # 模拟发送邮件 print(f"异步发送邮件: {event.data}") # 使用异步事件总线 async def main(): event_bus = AsyncEventBus() event_bus.subscribe('user.created', AsyncEmailHandler()) event = Event('user.created', {'email': 'john@example.com'}) await event_bus.publish(event) asyncio.run(main())

2.2 使用队列实现事件解耦

import asyncio from queue import Queue from threading import Thread class QueueEventBus: def __init__(self): self.queue = Queue() self.running = False self.thread = None def start(self): self.running = True self.thread = Thread(target=self._process_events) self.thread.start() def stop(self): self.running = False self.queue.put(None) # 发送终止信号 self.thread.join() def publish(self, event): self.queue.put(event) def _process_events(self): while self.running: event = self.queue.get() if event is None: break # 处理事件... print(f"处理事件: {event.event_type}") # 使用队列事件总线 event_bus = QueueEventBus() event_bus.start() event_bus.publish(Event('order.created', {'order_id': 1})) event_bus.publish(Event('order.paid', {'order_id': 1})) event_bus.stop()

三、事件驱动框架

3.1 使用RxPy响应式编程

from rx import Observable, operators as op # 创建事件流 events = Observable.from_iterable([ Event('click', {'x': 10, 'y': 20}), Event('click', {'x': 30, 'y': 40}), Event('scroll', {'offset': 100}) ]) # 过滤和处理事件 events.pipe( op.filter(lambda e: e.event_type == 'click'), op.map(lambda e: e.data) ).subscribe( on_next=lambda data: print(f"点击位置: {data}"), on_error=lambda e: print(f"错误: {e}"), on_completed=lambda: print("完成") )

3.2 使用Django信号

from django.dispatch import Signal, receiver # 定义信号 user_registered = Signal() # 注册信号处理器 @receiver(user_registered) def send_welcome_email(sender, **kwargs): user = kwargs['user'] print(f"发送欢迎邮件给: {user.email}") # 发送信号 user_registered.send(sender=None, user=User(email='john@example.com'))

四、生产环境实践

4.1 事件溯源模式

class EventStore: """事件存储""" def __init__(self): self.events = [] def append(self, event): self.events.append(event) def get_events_for_aggregate(self, aggregate_id): return [e for e in self.events if e.data.get('aggregate_id') == aggregate_id] class UserAggregate: def __init__(self, user_id): self.user_id = user_id self.name = None self.email = None def apply(self, event): if event.event_type == 'user.created': self.name = event.data['name'] self.email = event.data['email'] elif event.event_type == 'user.updated': if 'name' in event.data: self.name = event.data['name'] # 重建聚合状态 event_store = EventStore() events = event_store.get_events_for_aggregate(user_id=1) user = UserAggregate(user_id=1) for event in events: user.apply(event)

4.2 事件驱动微服务通信

import pika class RabbitMQEventBus: def __init__(self, host='localhost'): self.connection = pika.BlockingConnection(pika.ConnectionParameters(host)) self.channel = self.connection.channel() def publish(self, exchange, routing_key, event): self.channel.basic_publish( exchange=exchange, routing_key=routing_key, body=json.dumps(event.__dict__) ) def subscribe(self, queue, callback): def wrapper(ch, method, properties, body): event_data = json.loads(body) event = Event(event_data['event_type'], event_data['data']) callback(event) self.channel.basic_consume(queue=queue, on_message_callback=wrapper) self.channel.start_consuming()

五、事件驱动架构模式

5.1 事件队列模式

class EventQueue: def __init__(self): self.queue = [] self.processing = False def enqueue(self, event): self.queue.append(event) if not self.processing: self._process_next() def _process_next(self): if not self.queue: self.processing = False return self.processing = True event = self.queue.pop(0) # 异步处理事件 asyncio.create_task(self._handle_event(event)) async def _handle_event(self, event): try: # 处理事件 print(f"处理事件: {event.event_type}") finally: self._process_next()

5.2 事件调度器

import heapq class EventScheduler: def __init__(self): self.events = [] self.running = False def schedule(self, event, delay): """延迟delay秒后触发事件""" trigger_time = time.time() + delay heapq.heappush(self.events, (trigger_time, event)) async def start(self): self.running = True while self.running: if self.events: trigger_time, event = self.events[0] now = time.time() if now >= trigger_time: heapq.heappop(self.events) # 处理事件 print(f"触发延迟事件: {event.event_type}") else: await asyncio.sleep(trigger_time - now) else: await asyncio.sleep(0.1)

六、性能优化与监控

6.1 事件处理监控

from dataclasses import dataclass from time import time @dataclass class EventMetrics: event_type: str count: int = 0 total_time: float = 0 min_time: float = float('inf') max_time: float = 0 class EventMonitor: def __init__(self): self.metrics = defaultdict(EventMetrics) def record(self, event_type, duration): metric = self.metrics[event_type] metric.count += 1 metric.total_time += duration metric.min_time = min(metric.min_time, duration) metric.max_time = max(metric.max_time, duration) def report(self): for event_type, metric in self.metrics.items(): avg_time = metric.total_time / metric.count if metric.count > 0 else 0 print(f"{event_type}: {metric.count}次, 平均{avg_time:.3f}s") # 使用监控 monitor = EventMonitor() def handle_event(event): start = time() # 处理事件 time.sleep(0.1) monitor.record(event.event_type, time() - start)