feat: sse && multi_merchant completed

2025-11-11 10:56:20 +08:00
parent b2078dfa46
commit 2aafb93aad
43 changed files with 2445 additions and 244 deletions
--- a/sse/README.md
+++ b/sse/README.md
@@ -0,0 +1,154 @@
+# SSE (Server-Sent Events) 模块
+
+基于 Django REST Framework 的服务器推送事件实现。
+
+## 功能特性
+
+- ✅ 使用 DRF 处理请求和响应
+- ✅ 支持多种请求格式（JSON、Form Data、Multipart）
+- ✅ 自动数据验证和序列化
+- ✅ 异步支持，高并发处理
+- ✅ 心跳机制保持连接活跃
+- ✅ 自动清理断开的连接
+- ✅ 连接状态监控
+
+## API 端点
+
+### 1. 订阅 SSE 事件流
+
+**端点**: `GET /sse/`
+
+客户端连接此端点保持长连接，接收服务器推送的事件。
+
+**示例**:
+```bash
+curl -N http://localhost:8000/sse/
+```
+
+**JavaScript 示例**:
+```javascript
+const eventSource = new EventSource('http://localhost:8000/sse/');
+
+eventSource.onmessage = function(event) {
+    const data = JSON.parse(event.data);
+    console.log('收到消息:', data);
+};
+```
+
+---
+
+### 2. 推送事件到所有客户端
+
+**端点**: `POST /sse/push/`
+
+向所有已连接的客户端广播消息。
+
+**请求参数**:
+- `message` (必填): 消息内容
+- `type` (可选): 事件类型，默认为 'message'
+
+**支持的请求格式**:
+
+#### JSON 格式
+```bash
+curl -X POST http://localhost:8000/sse/push/ \
+  -H "Content-Type: application/json" \
+  -d '{"message": "Hello, SSE!", "type": "notification"}'
+```
+
+#### Form Data 格式
+```bash
+curl -X POST http://localhost:8000/sse/push/ \
+  -d "message=Hello, SSE!" \
+  -d "type=notification"
+```
+
+#### Multipart Form Data
+```bash
+curl -X POST http://localhost:8000/sse/push/ \
+  -F "message=Hello, SSE!" \
+  -F "type=notification"
+```
+
+**响应示例**:
+```json
+{
+    "status": "success",
+    "message": "Event sent to 3 client(s)",
+    "clients": 3,
+    "sent": 3
+}
+```
+
+---
+
+### 3. 获取连接状态
+
+**端点**: `GET /sse/status/`
+
+查询当前 SSE 服务器的状态和连接数。
+
+**示例**:
+```bash
+curl http://localhost:8000/sse/status/
+```
+
+**响应示例**:
+```json
+{
+    "status": "running",
+    "clients": 3,
+    "message": "SSE server is running with 3 active connection(s)"
+}
+```
+
+## 启动服务器
+
+使用 Uvicorn (ASGI 服务器) 启动：
+
+```bash
+# 开发环境
+uvicorn flower.asgi:application --reload --host 0.0.0.0 --port 8000
+
+# 生产环境
+uvicorn flower.asgi:application --host 0.0.0.0 --port 8000 --workers 4
+```
+
+## 测试
+
+打开 `sse_test.html` 在浏览器中测试：
+
+1. 点击"连接 SSE"建立连接
+2. 输入消息
+3. 点击"发送 (JSON)"或"发送 (Form Data)"测试不同格式
+4. 点击"获取连接状态"查看当前连接数
+5. 打开多个浏览器标签测试广播功能
+
+## Python 客户端示例
+
+```python
+import requests
+import sseclient  # pip install sseclient-py
+
+# 订阅事件
+response = requests.get('http://localhost:8000/sse/', stream=True)
+client = sseclient.SSEClient(response)
+
+for event in client.events():
+    print(f'收到消息: {event.data}')
+```
+
+## 技术实现
+
+- **异步视图**: 使用 `async def` 实现异步处理
+- **队列机制**: 每个连接对应一个 `asyncio.Queue`
+- **心跳**: 30 秒超时，自动发送心跳保持连接
+- **DRF 集成**: 使用 DRF 的 `@api_view` 和序列化器
+- **多格式支持**: 自动解析 JSON、Form Data、Multipart 等格式
+
+## 注意事项
+
+1. 必须使用 ASGI 服务器（如 Uvicorn、Daphne）运行
+2. 不支持使用传统的 WSGI 服务器（如 Gunicorn + WSGI）
+3. 如果使用 Nginx，需要禁用缓冲：`X-Accel-Buffering: no`
+4. SSE 使用 GET 请求，注意 CORS 配置
--- a/sse/init.py
+++ b/sse/init.py
@@ -0,0 +1,19 @@
+from django.apps import AppConfig
+from . import services
+import signal, sys
+
+
+class SseConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'sse'
+
+    def ready(self):
+        def cleanup_on_shutdown():
+            """在接收到终止信号时设置关闭事件"""
+            from .views import _shutdown_event
+            _shutdown_event.set()
+            services.cleanup_all_connections()
+            sys.exit(0)
+
+        signal.signal(signal.SIGINT, lambda s, f: cleanup_on_shutdown())
+        signal.signal(signal.SIGTERM, lambda s, f: cleanup_on_shutdown())
--- a/sse/minimal_sse.py
+++ b/sse/minimal_sse.py
@@ -0,0 +1,96 @@
+"""
+极简SSE实现 - 专注于快速关闭
+关键原则：
+1. 不使用任何全局状态或复杂的队列系统
+2. 使用最短的超时时间
+3. 依靠HTTP连接的自然断开机制
+"""
+import json
+import time
+import signal
+import threading
+from django.http import HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+# 简单的全局关闭标志
+shutdown_requested = threading.Event()
+
+def trigger_shutdown():
+    """触发全局关闭"""
+    shutdown_requested.set()
+
+def is_shutdown_requested():
+    """检查是否请求关闭"""
+    return shutdown_requested.is_set()
+
+@csrf_exempt 
+def minimal_sse_view(request):
+    """极简SSE视图 - 专注于快速响应关闭信号"""
+    
+    if request.method == 'OPTIONS':
+        response = HttpResponse()
+        origin = request.META.get('HTTP_ORIGIN')
+        if origin:
+            response['Access-Control-Allow-Origin'] = origin
+            response['Access-Control-Allow-Methods'] = 'GET, OPTIONS'
+            response['Access-Control-Allow-Headers'] = 'authorization, content-type'
+            response['Access-Control-Allow-Credentials'] = 'true'
+        return response
+    
+    def quick_stream():
+        """快速响应的流生成器"""
+        try:
+            # 发送连接消息
+            yield f"data: {json.dumps({'type': 'connected', 'time': time.time()})}\n\n"
+            
+            # 极短循环 - 每0.2秒检查关闭信号
+            counter = 0
+            while not shutdown_requested.is_set():
+                counter += 1
+                
+                # 每5次循环(1秒)发送心跳
+                if counter % 5 == 0:
+                    yield f": heartbeat\n\n"
+                
+                # 非常短的睡眠，快速响应关闭
+                time.sleep(0.2)
+                
+                # 最多运行100次循环(20秒)后自动断开，防止永久阻塞
+                if counter > 100:
+                    break
+            
+            # 发送关闭消息
+            yield f"data: {json.dumps({'type': 'closing'})}\n\n"
+            
+        except GeneratorExit:
+            pass
+        except Exception as e:
+            print(f"SSE stream error: {e}")
+    
+    response = HttpResponse(quick_stream(), content_type='text/event-stream')
+    response['Cache-Control'] = 'no-cache, no-store'
+    response['Connection'] = 'close'  # 明确告诉客户端这是短连接
+    
+    origin = request.META.get('HTTP_ORIGIN')
+    if origin:
+        response['Access-Control-Allow-Origin'] = origin
+        response['Access-Control-Allow-Credentials'] = 'true'
+    
+    return response
+
+
+# 注册简单的信号处理器
+def setup_minimal_handlers():
+    def handle_shutdown(signum, frame):
+        print(f"Minimal SSE: received signal {signum}")
+        trigger_shutdown()
+    
+    signal.signal(signal.SIGTERM, handle_shutdown)
+    signal.signal(signal.SIGINT, handle_shutdown)
+
+# 立即设置处理器
+try:
+    setup_minimal_handlers()
+    print("Minimal SSE handlers registered")
+except:
+    pass
--- a/sse/serializers.py
+++ b/sse/serializers.py
@@ -0,0 +1,27 @@
+from rest_framework import serializers
+
+
+class PushSSEEventSerializer(serializers.Serializer):
+    """
+    SSE 事件推送序列化器
+    
+    用于验证推送到 SSE 客户端的事件数据
+    """
+    message = serializers.CharField(
+        required=True,
+        help_text="要发送的消息内容",
+        max_length=10000,
+        allow_blank=False,
+    )
+    type = serializers.CharField(
+        required=False,
+        default='message',
+        help_text="事件类型，如：message, notification, alert 等",
+        max_length=100,
+    )
+    
+    def validate_message(self, value):
+        """验证消息内容"""
+        if not value or not value.strip():
+            raise serializers.ValidationError("消息内容不能为空")
+        return value.strip()
--- a/sse/services.py
+++ b/sse/services.py
@@ -0,0 +1,96 @@
+import queue
+import logging
+
+logger = logging.getLogger(__name__)
+
+# 存储所有活动的 SSE 连接队列（同步队列）
+_connections = set()
+
+
+def get_active_connections():
+    """获取当前所有活动的 SSE 连接队列"""
+    return _connections
+
+
+def push_connection(conn_queue):
+    """
+    将一个新的连接队列添加到活动连接集合中
+    
+    参数:
+    - conn_queue: queue.Queue 实例
+    """
+    _connections.add(conn_queue)
+    logger.info(f"新的 SSE 连接建立，当前连接数: {len(_connections)}")
+
+
+def remove_connection(conn_queue):
+    """
+    从活动连接集合中移除一个连接队列
+    
+    参数:
+    - conn_queue: queue.Queue 实例
+    """
+    _connections.discard(conn_queue)
+    logger.info(f"SSE 连接断开，当前连接数: {len(_connections)}")
+
+
+def cleanup_all_connections():
+    """
+    清理所有连接（用于服务器关闭时）
+    """
+    connections_list = list(_connections)
+    for conn_queue in connections_list:
+        try:
+            # 发送关闭信号到队列
+            if hasattr(conn_queue, 'put_nowait'):
+                conn_queue.put_nowait({'type': 'server_shutdown', 'message': 'Server shutting down'})
+                logger.debug(f'shutdown connection {conn_queue}')
+        except Exception:
+            pass  # 忽略错误，因为连接可能已经断开
+    _connections.clear()
+    logger.info("所有SSE连接已清理")
+
+
+def push_sse_event_to_all(event_data: dict):
+    """
+    向所有连接的客户端广播一个 SSE 事件（同步队列版本）
+    
+    参数:
+    - event_data: 要发送的事件数据（字典）
+    """
+    disconnected = []
+    
+    for conn_queue in list(_connections):
+        try:
+            # 使用put_nowait非阻塞发送消息
+            conn_queue.put_nowait(event_data)
+        except queue.Full:
+            # 队列满了，说明客户端处理消息太慢，跳过这条消息
+            logger.warning(f"队列已满，跳过消息: {event_data.get('type', 'unknown')}")
+        except Exception as e:
+            # 其他异常可能表示连接已断开
+            logger.error(f"向队列发送消息失败: {e}")
+            disconnected.append(conn_queue)
+    
+    # 清理断开的连接
+    for conn_queue in disconnected:
+        remove_connection(conn_queue)
+
+
+def push_simple_message_with_object_id(event_type: str, message: str, object_id):
+    """
+    向所有连接的客户端广播一个简单消息事件，包含关联对象ID（WSGI同步版本）
+    
+    参数:
+    - event_type: 事件类型字符串
+    - message: 消息内容字符串
+    - object_id: 关联对象的ID（整数或字符串）
+    """
+    event_data = {
+        'mode': 'simple_message',
+        'type': event_type,
+        'message': message,
+        'object_id': object_id,
+    }
+    push_sse_event_to_all(event_data)
+    logger.info(f"广播 SSE 事件: type={event_type}, object_id={object_id}, 接收客户端数={len(_connections)}")
--- a/sse/simple_sse.py
+++ b/sse/simple_sse.py
@@ -0,0 +1,154 @@
+"""
+简化的SSE实现，专门解决uvicorn无法退出的问题
+"""
+import json
+import time
+import threading
+from collections import defaultdict
+from django.http import HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+
+# 全局状态
+_clients = {}  # 存储客户端连接的状态
+_client_counter = 0
+_shutdown_flag = threading.Event()
+_lock = threading.Lock()
+
+
+def get_next_client_id():
+    """获取下一个客户端ID"""
+    global _client_counter
+    with _lock:
+        _client_counter += 1
+        return _client_counter
+
+
+def add_client(client_id):
+    """添加客户端"""
+    with _lock:
+        _clients[client_id] = {
+            'active': True,
+            'messages': [],
+            'last_heartbeat': time.time()
+        }
+
+
+def remove_client(client_id):
+    """移除客户端"""
+    with _lock:
+        _clients.pop(client_id, None)
+
+
+def broadcast_message(message):
+    """广播消息给所有客户端"""
+    with _lock:
+        for client_id, client_data in _clients.items():
+            if client_data['active']:
+                client_data['messages'].append(message)
+
+
+def get_client_messages(client_id):
+    """获取客户端的消息"""
+    with _lock:
+        if client_id in _clients:
+            messages = _clients[client_id]['messages'][:]
+            _clients[client_id]['messages'].clear()
+            _clients[client_id]['last_heartbeat'] = time.time()
+            return messages
+    return []
+
+
+def cleanup_clients():
+    """清理所有客户端"""
+    global _shutdown_flag
+    _shutdown_flag.set()
+    with _lock:
+        for client_data in _clients.values():
+            client_data['active'] = False
+        _clients.clear()
+
+
+@csrf_exempt
+def simple_sse_view(request):
+    """简化的SSE视图"""
+    if request.method == 'OPTIONS':
+        response = HttpResponse()
+        origin = request.META.get('HTTP_ORIGIN')
+        if origin:
+            response['Access-Control-Allow-Origin'] = origin
+            response['Access-Control-Allow-Methods'] = 'GET, OPTIONS'
+            response['Access-Control-Allow-Headers'] = 'authorization, content-type, cache-control, accept'
+            response['Access-Control-Allow-Credentials'] = 'true'
+            response['Access-Control-Max-Age'] = '86400'
+        return response
+
+    def event_generator():
+        client_id = get_next_client_id()
+        add_client(client_id)
+        
+        try:
+            # 发送连接成功消息
+            yield f"data: {json.dumps({'type': 'connected', 'client_id': client_id})}\n\n"
+            
+            # 主循环 - 使用更短的检查间隔
+            iterations = 0
+            while not _shutdown_flag.is_set():
+                # 每5次迭代检查一次消息（约0.5秒）
+                if iterations % 5 == 0:
+                    messages = get_client_messages(client_id)
+                    for message in messages:
+                        yield f"data: {json.dumps(message)}\n\n"
+                
+                # 每50次迭代发送心跳（约5秒）
+                if iterations % 50 == 0:
+                    yield f": heartbeat {time.time()}\n\n"
+                
+                # 短暂睡眠，让出CPU并允许快速响应关闭信号
+                time.sleep(0.1)
+                iterations += 1
+                
+                # 如果关闭标志被设置，退出循环
+                if _shutdown_flag.is_set():
+                    break
+            
+            # 发送关闭消息
+            yield f"data: {json.dumps({'type': 'shutdown', 'message': 'Server shutting down'})}\n\n"
+            
+        except Exception as e:
+            print(f"SSE error for client {client_id}: {e}")
+        finally:
+            remove_client(client_id)
+            print(f"Client {client_id} disconnected")
+    
+    response = HttpResponse(event_generator(), content_type='text/event-stream')
+    response['Cache-Control'] = 'no-cache'
+    response['X-Accel-Buffering'] = 'no'
+    
+    # CORS headers
+    origin = request.META.get('HTTP_ORIGIN')
+    if origin:
+        response['Access-Control-Allow-Origin'] = origin
+        response['Access-Control-Allow-Credentials'] = 'true'
+    
+    return response
+
+
+def broadcast_test_message():
+    """广播测试消息"""
+    message = {
+        'type': 'test',
+        'message': 'Test message',
+        'timestamp': time.time()
+    }
+    broadcast_message(message)
+    return len(_clients)
+
+
+def get_status():
+    """获取状态"""
+    with _lock:
+        return {
+            'clients': len(_clients),
+            'shutdown': _shutdown_flag.is_set()
+        }
--- a/sse/views.py
+++ b/sse/views.py
@@ -1,45 +1,131 @@
-from django.http.response import StreamingHttpResponse
+from django.http import StreamingHttpResponse, HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+from django.views.decorators.http import require_http_methods
 from rest_framework.decorators import api_view, permission_classes
+from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
-from contextlib import suppress
-import asyncio
+from rest_framework import status
+from . import services
+import queue
+import json


-_connections = set()
-
-
-@api_view(['POST'])
-@permission_classes([])
+@csrf_exempt
+@require_http_methods(["GET", "OPTIONS"])
 def create_sse_event(request):
    """
-    创建一个简单的SSE事件流响应，用于测试和演示目的。
+    创建一个 SSE 事件流响应。
+    客户端连接到此端点后会保持长连接，等待服务器推送事件。
+    
+    使用方法：
+    - GET /sse/
+    - 保持连接打开以接收实时事件
+    
+    注意：
+    - 使用纯 Django 视图，不使用 DRF，避免内容协商导致的 406 错误
+    - SSE 需要特殊的 CORS 配置
    """
-
-    def sse_stream():
-        queue = asyncio.Queue()
-        _connections.add(queue)
-
+    # 处理 OPTIONS 预检请求
+    if request.method == 'OPTIONS':
+        response = HttpResponse()
+        origin = request.META.get('HTTP_ORIGIN')
+        if origin:
+            response['Access-Control-Allow-Origin'] = origin
+            response['Access-Control-Allow-Methods'] = 'GET, OPTIONS'
+            response['Access-Control-Allow-Headers'] = 'authorization, content-type, cache-control, accept'
+            response['Access-Control-Allow-Credentials'] = 'true'
+            response['Access-Control-Max-Age'] = '86400'  # 24小时
+        return response
+    
+    def event_stream():
+        # 创建一个同步队列用于接收消息
+        conn_queue = queue.Queue(maxsize=100)
+        services.push_connection(conn_queue)
+        
        try:
+            # 发送初始连接成功消息
+            yield f"data: {json.dumps({'type': 'connected', 'message': 'SSE connection established'})}\n\n"
+            
+            # 持续从队列中获取消息并发送给客户端
            while True:
-                data = queue.get()
-                yield f"data: {data}\n\n"
+                try:
+                    # 使用同步方式等待新消息，带超时（30秒）以便发送心跳
+                    message = conn_queue.get(timeout=30.0)
+                    yield f"data: {json.dumps(message)}\n\n"
+                except queue.Empty:
+                    # 30秒超时，发送心跳保持连接活跃
+                    yield f": heartbeat\n\n"
+                except Exception as e:
+                    print(f"Error in SSE stream: {e}")
+                    break
        finally:
-            _connections.remove(queue)
-
-    response = StreamingHttpResponse(sse_stream(), content_type='text/event-stream')
+            # 清理：从连接集合中移除此队列
+            services.remove_connection(conn_queue)
+    
+    response = StreamingHttpResponse(
+        event_stream(),
+        content_type='text/event-stream',
+    )
+    
+    # SSE 必需的响应头
    response['Cache-Control'] = 'no-cache'
+    response['X-Accel-Buffering'] = 'no'  # 禁用 nginx 缓冲
+    
+    # CORS 响应头（django-cors-headers 中间件会自动添加，但我们显式设置以确保）
+    # 如果请求带有 Origin 头，手动添加 CORS 响应头
+    origin = request.META.get('HTTP_ORIGIN')
+    if origin:
+        response['Access-Control-Allow-Origin'] = origin
+        response['Access-Control-Allow-Credentials'] = 'true'
+    
    return response


@api_view(['POST'])
-@permission_classes([])
-def push_sse_event(request):
+@permission_classes([AllowAny])
+def push_test_event(request):
    """
-    向所有连接的客户端广播一个SSE事件。
-    请求体应包含一个 'message' 字段，表示要发送的消息内容。
+    向所有连接的客户端广播一个 SSE 测试事件。
    """
-    message = request.data.get('message', 'Hello, SSE!')
-    for q in list(_connections):
-        with suppress(asyncio.QueueFull):
-            q.put_nowait(message)
-    return Response({'status': 'message sent'})
+    services.push_simple_message_with_object_id('order_paid', '订单已支付', 12345)
+    return Response({
+        'status': 'ok',
+        'message': 'Test event broadcasted',
+        'clients': len(services.get_active_connections())
+    })
+
+
+@api_view(['GET'])
+@permission_classes([AllowAny])
+def get_sse_status(request):
+    """
+    获取 SSE 连接状态信息。
+    
+    返回：
+    - clients: 当前连接的客户端数量
+    - status: 服务状态
+    """
+    active_connections = services.get_active_connections()
+    return Response({
+        'status': 'running',
+        'clients': len(active_connections),
+        'message': f'SSE server is running with {len(active_connections)} active connection(s)',
+    })
+
+
+@api_view(['POST'])
+@permission_classes([AllowAny])
+def shutdown_sse(request):
+    """
+    优雅关闭所有SSE连接的端点
+    """
+    services.push_sse_event_to_all({
+        'type': 'server_shutdown',
+        'message': 'Server is shutting down, please reconnect later'
+    })
+    
+    return Response({
+        'status': 'ok',
+        'message': 'Shutdown signal sent to all SSE connections',
+        'clients': len(services.get_active_connections())
+    })