Django事件驱动架构怎么落地

wen python案例 2026-07-12 1

本文目录导读：

Django事件驱动架构怎么落地

核心思路：Event Bus + 异步 Worker
基础设施选型建议
落地步骤与代码示例
落地步骤

落地 Django 事件驱动架构，核心思路是将业务逻辑中的“状态变更”或“动作发生”抽象为事件，通过异步解耦的方式，让事件的生产者和消费者各自独立演进。

在 Django 这个同步阻塞的 WSGI 框架中落地，需要结合第三方消息中间件和异步任务系统,下面是一个从架构设计到代码实现的分步方案。

核心思路：Event Bus + 异步 Worker

事件定义：用 Python 类（Pydantic/Dataclass）明确事件的 type、payload、metadata。
事件发布：在 Django 的业务逻辑中（如 Model 的 save() 信号或 View 中），不直接调用后续逻辑，而是向 Event Bus（消息队列）发送事件消息。
事件存储与路由：使用 Redis Streams、RabbitMQ 或 Kafka 作为中间件。
事件处理：独立的异步 Worker 进程（Celery、Dramatiq、或基于 asyncio 的 consumer）监听队列，取出事件并调用对应的 Handler。

基础设施选型建议

场景	推荐中间件	理由
高吞吐、强持久化、顺序消费	Apache Kafka	日志、审计、大数据管道
常规业务解耦、低延迟、路由灵活	RabbitMQ	用户注册、通知、任务调度
简单、无外部依赖、小团队	Redis Streams	轻量级，与 Django 进程协作好
与 Celery 深度集成	Celery + RabbitMQ/Redis	复用已有的异步任务队列

落地步骤与代码示例

定义事件（统一契约）

创建一个 events.py，使用 Pydantic 进行强类型定义。

# events.py
from pydantic import BaseModel
from datetime import datetime
import uuid
class DomainEvent(BaseModel):
    event_id: str = str(uuid.uuid4())
    timestamp: datetime = datetime.utcnow()
    event_type: str
    payload: dict
# 具体事件
class UserRegisteredEvent(DomainEvent):
    event_type: str = "user.registered"
    payload: dict  # {'user_id': 1, 'email': '...'}
class OrderCreatedEvent(DomainEvent):
    event_type: str = "order.created"
    payload: dict  # {'order_id': 1001, 'user_id': 1, 'total': 99.9}

事件发布器（在 Django 中）

创建一个 event_publisher.py,负责将事件序列化并发送到消息队列。

# event_publisher.py
import json
import redis  # 假设使用 Redis Streams
from django.conf import settings
redis_client = redis.Redis.from_url(settings.REDIS_URL)
class EventPublisher:
    @staticmethod
    def publish(event: DomainEvent):
        """将事件发送到 Redis Stream"""
        event_data = event.model_dump_json()  # 序列化为 JSON
        # 使用 Stream 的 'evts' 键，自动生成消息ID
        redis_client.xadd("stream:domain_events", {"event": event_data})
        # 也可以发布到不同的 Stream（按事件类型）
        # redis_client.xadd(f"stream:{event.event_type}", {"event": event_data})

在业务代码中发出事件

在 View 或 Service 层显式发布（推荐）

# views.py
from events import UserRegisteredEvent, EventPublisher
def register_user(request):
    user = User.objects.create(...)  # 创建用户
    # 发布事件
    event = UserRegisteredEvent(payload={"user_id": user.id, "email": user.email})
    EventPublisher.publish(event)
    return JsonResponse({"status": "ok"})

利用 Django Signal 发布（适合旧项目改造）

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from events import UserRegisteredEvent, EventPublisher
@receiver(post_save, sender=User)
def user_saved_handler(sender, instance, created, **kwargs):
    if created:  # 仅当创建时发事件
        event = UserRegisteredEvent(payload={"user_id": instance.id, "email": instance.email})
        EventPublisher.publish(event)

事件处理器（异步 Worker）

建立一个独立的 Python 进程（event_worker.py），持续监听队列并调用 handler。

# event_worker.py
import json
import redis
from handlers import handle_user_registered, handle_order_created
redis_client = redis.Redis.from_url("redis://localhost:6379/0")
# 事件类型 -> Handler 映射表
EVENT_HANDLERS = {
    "user.registered": [handle_user_registered],  # 一个事件可以有多个 handler
    "order.created": [handle_order_created, handle_send_notification],
}
def process_event(event_data: dict):
    event_type = event_data.get("event_type")
    handlers = EVENT_HANDLERS.get(event_type, [])
    for handler in handlers:
        try:
            handler(event_data)  # 执行具体的业务逻辑
        except Exception as e:
            print(f"Handler {handler.__name__} failed: {e}")
            # 可以重试或发送到死信队列
def listen():
    print("Worker listening for events...")
    while True:
        # 使用 Redis Stream 的阻塞读取
        results = redis_client.xread({"stream:domain_events": "0-0"}, count=10, block=5000)
        for stream, messages in results:
            for message_id, data in messages:
                event_dict = json.loads(data[b"event"].decode())
                process_event(event_dict)
                # 消费成功，确认消息
                redis_client.xdel("stream:domain_events", message_id)
if __name__ == "__main__":
    listen()

更好的方式：使用 Celery 作为 Worker

如果项目已经用了 Celery，可以直接用 Celery 的任务来处理事件，省去自己写 Worker 的麻烦。

# tasks.py (Celery)
from celery import shared_task
from handlers import handle_user_registered
@shared_task
def process_user_registered(event_payload):
    handle_user_registered(event_payload)
# 发布事件时改为调用异步任务
from tasks import process_user_registered
# EventPublisher.publish(event) 改为 ->
process_user_registered.delay(event.model_dump())

关键设计原则

Handler 幂等性：事件可能被重复消费（网络抖动、重试），Handler 需能处理重复（get_or_create 或检查唯一键）。
顺序消费：如果业务强依赖事件顺序（如先创建订单，再扣库存），使用 Kafka 分区 或 Redis Streams 单消费组，保证同一用户/订单的事件按序处理。
失败与重试：Handler 失败后不要丢弃，应重试，使用 死信队列 存储重试多次仍失败的事件。
事件溯源（可选）：将事件持久化保存到数据库,作为审计日志或状态恢复的依据。

落地步骤

选中间件：根据你的场景选 Kafka、RabbitMQ 或 Redis。
定义事件：写 Pydantic 类,明确类型和结构。
替换调用：在需要解耦的地方（发邮件、写日志、更新下游），改为只发事件,不直接调用。
启动 Worker：写一个独立的进程或 Celery Task，监听事件并调用 Handler。
渐进改造：不必一步到位，可以先从非关键链路（如发欢迎邮件、通知）开始实践,再逐步扩展到核心业务。

一个建议：在 Django 中落地事件驱动，最容易失败的点是 事务一致性问题，即：数据库操作成功了，但事件没发出去（或发了但 Handler 没执行）,建议方案：

Transaction Outbox 模式：先在同一数据库事务中将事件存入 outbox 表，再通过一个定时任务或 CDC（Change Data Capture）工具，将 outbox 中的事件可靠地投递到消息队列，这是金融级推荐的方案，能保证“数据库变更”与“事件发布”的原子性。