本文详解如何在 django 应用中实现「独立运行的异步爬虫 → 持续产出数据 → 用户页面自动实时更新」的完整链路，推荐采用 redis 作为中间存储 + server-sent events（sse）作为前端推送通道，并规避 asyncio 嵌套事件循环等典型陷阱。
本文详解如何在 django 应用中实现「独立运行的异步爬虫 → 持续产出数据 → 用户页面自动实时更新」的完整链路，推荐采用 redis 作为中间存储 + server-sent events（sse）作为前端推送通道，并规避 asyncio 嵌套事件循环等典型陷阱。
在构建实时数据看板类应用（如监控仪表盘、行情聚合页、舆情采集面板）时，一个常见且关键的需求是：后台爬虫长期、异步、独立运行；前端用户访问页面后，无需轮询或手动刷新，即可持续接收最新数据流。你提出的三种技术路径（Multiprocessing Pipe、WebSocket、SSE）各有适用场景，但结合 Django 生态与生产稳定性，Redis + SSE 是当前最轻量、可靠且符合分层设计原则的方案——它解耦了数据生产与消费，规避了线程/协程生命周期冲突，也避免了 WebSocket 的双向复杂性与连接管理开销。
✅ 推荐架构：生产者-消费者分离 + Redis 中间件
你的最终选择（“用 Redis 存储爬虫结果，Django 视图读取”）本质上是正确的，但需进一步升级为实时推送模式，而非仅静态拉取。完整流程如下：
强戈环形山: q15040409-刻画家功能概述强戈环形山: q15040409-刻画家是一项面向实际任务的技能，主要用于平顶山: 维西翁: 1:1.0. 塔格斯: rap.github: 自动化。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。从功能定位来看，该
-
异步爬虫（独立进程）
使用 asyncio + httpx/aiohttp 编写爬虫脚本（如 scraper.py），每完成一轮采集+清洗后，将结构化数据（如 JSON）写入 Redis 的 STRING 或 HASH 键中，并可选地触发 PUBLISH 通知：# scraper.py（独立运行：python scraper.py） import asyncio import json import redis from httpx import AsyncClient r = redis.Redis(host="localhost", port=6379, db=0) async def fetch_and_store(): async with AsyncClient() as client: resp = await client.get("https://api.example.com/data") data = resp.json() # 写入最新快照（覆盖式） r.set("latest_scraped_data", json.dumps(data)) # 可选：发布更新事件（供其他服务监听） r.publish("scraping:updated", "data_refreshed") async def main(): while True: try: await fetch_and_store() except Exception as e: print(f"Scraping error: {e}") await asyncio.sleep(30) # 每30秒执行一次 if __name__ == "__main__": asyncio.run(main()) # ✅ 独立事件循环，无嵌套风险 -
Django SSE 视图（ASGI 模式下）
在 Django 中创建一个支持长连接的 SSE 视图。关键点：- 必须部署在 ASGI 服务器（如 Daphne、Uvicorn）下；
- 使用 StreamingHttpResponse 返回 text/event-stream；
- 利用 Redis 的 SUBSCRIBE 或轮询 GET 实现低延迟更新（推荐轮询，更稳定）；
- 严禁在 ASGI 视图中调用 asyncio.run()（这正是你遇到 RuntimeError 的原因——Django ASGI 已在运行事件循环）。
# views.py import json import time import redis from django.http import StreamingHttpResponse from django.views.decorators.csrf import csrf_exempt r = redis.Redis(host="localhost", port=6379, db=0) def event_stream(): last_data = None while True: # 轮询获取最新数据（轻量，避免 SUBSCRIBE 复杂性） current_data = r.get("latest_scraped_data") if current_data and current_data != last_data: last_data = current_data yield f"data: {current_data.decode()}\n\n" time.sleep(1) # 防止过度轮询 @csrf_exempt def sse_data_view(request): response = StreamingHttpResponse( event_stream(), content_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) response["Access-Control-Allow-Origin"] = "*" # 如需跨域 return response -
前端自动订阅（HTML + JavaScript）
页面加载后建立 SSE 连接，自动处理数据更新：<!-- template.html --> <div id="data-display">Loading...</div> <script> const eventSource = new EventSource("/sse-data/"); eventSource.onmessage = function(event) { const data = JSON.parse(event.data); document.getElementById("data-display").textContent = `Updated at ${new Date().toLocaleTimeString()}: ${data.value || 'N/A'}`; }; eventSource.onerror = function() { console.error("SSE connection lost"); }; </script>
⚠️ 关键注意事项与避坑指南
- 不要混用 WSGI 与 SSE：WSGI 服务器（如 Gunicorn）不支持长连接流式响应，必须使用 ASGI（配置 ASGI_APPLICATION = "myproject.asgi.application"）。
- Redis 连接复用：在 Django 中初始化全局 redis.Redis() 实例（非每次请求新建），避免连接泄漏。
- 爬虫健壮性：为爬虫添加错误重试、日志记录、崩溃自重启（如用 systemd 或 supervisord 守护）。
-
替代方案对比：
- ✅ Redis + SSE：简单、低延迟、无状态、易扩展，适合单向广播场景；
- ⚠️ Django Channels + WebSocket：适合需要用户鉴权、房间隔离、双向交互的场景（如聊天室），但开发与运维成本更高；
- ❌ Multiprocessing Pipe：进程间通信难以跨机器、无法水平扩展，且 Django 主进程无法安全持有子进程句柄。
总结
你已找到核心解法——用 Redis 解耦爬虫与 Web 层。在此基础上，通过 SSE 将“被动查询”升级为“主动推送”，即可完美满足「用户访问即实时订阅」的需求。整个方案完全兼容 Django 3.1+ 的异步生态，无需修改 ORM 或引入 Celery/Channels 等重型组件，兼顾开发效率与生产稳定性。下一步建议增加数据版本号（如 Redis INCR 计数器）或 Last-Event-ID 支持断线续传，让系统更加健壮。
