From 45c58f757c4388ec58e78db8bf15bac9cc43369c Mon Sep 17 00:00:00 2001 From: colaftc Date: Fri, 9 Jan 2026 18:00:22 +0800 Subject: [PATCH] fix: added has_started into printing jobs list srz --- api_v1/views/printing/serializers.py | 3 +- docs/sse_refactor.md | 909 +++++++++++++++++++++++++++ 2 files changed, 911 insertions(+), 1 deletion(-) create mode 100644 docs/sse_refactor.md diff --git a/api_v1/views/printing/serializers.py b/api_v1/views/printing/serializers.py index 6a50480..b338501 100644 --- a/api_v1/views/printing/serializers.py +++ b/api_v1/views/printing/serializers.py @@ -279,6 +279,7 @@ class PrintingJobListSerializer(serializers.ModelSerializer): product_image_url = serializers.SerializerMethodField() status = serializers.CharField(read_only=True) work_state_display = serializers.SerializerMethodField() + has_started = serializers.BooleanField(read_only=True) is_completed = serializers.BooleanField(read_only=True) business_object_id = serializers.SerializerMethodField() batch_advance_records = serializers.SerializerMethodField() @@ -287,7 +288,7 @@ class PrintingJobListSerializer(serializers.ModelSerializer): model = models.PrintingJob fields = [ 'id', 'original_id', 'printing_order', 'printing_order_id', 'product', 'product_name', - 'product_image_url', + 'product_image_url', 'has_started', 'quantity', 'unit', 'size', 'pieces', 'description', 'work_state', 'work_state_display', 'status', 'is_completed', 'business_object_id', diff --git a/docs/sse_refactor.md b/docs/sse_refactor.md new file mode 100644 index 0000000..9fc50ed --- /dev/null +++ b/docs/sse_refactor.md @@ -0,0 +1,909 @@ +# SSE 模块重构文档 + +## 1. 背景与问题 + +### 1.1 当前状态 + +SSE（Server-Sent Events）模块已被停用（返回 405），原因是出现了**致命的数据库连接泄漏问题**： + +- Django 的 SSE 长连接会无限制地消耗数据库连接（Database Connection） +- 连接长时间不释放，导致数据库连接池耗尽 +- 严重影响系统稳定性，导致其他业务无法正常访问数据库 + +### 1.2 重构目标 + +将 SSE 模块迁移到**独立的 Golang 进程**运行，与 Django 后端解耦： + +- Golang 的 goroutine 模型天然适合处理大量长连接 +- 不依赖 Django 的数据库连接池 +- 更好的资源管理和并发性能 +- 系统解耦，SSE 服务可独立扩展和部署 + +--- + +## 2. 当前架构概述 + +### 2.1 目录结构 + +``` +sse/ +├── __init__.py +├── admin.py +├── apps.py +├── auth_utils.py # JWT 认证工具 +├── client_example.js # 前端客户端示例 +├── migrations/ +├── minimal_sse.py # 极简实现（尝试解决关闭问题） +├── models.py # 空模型文件 +├── serializers.py # DRF 序列化器 +├── services.py # 核心服务层（连接管理、消息推送） +├── simple_sse.py # 简化实现（尝试解决关闭问题） +└── views.py # API 视图（已禁用） +``` + +### 2.2 当前 API 端点 + +| 端点 | 方法 | 描述 | 状态 | +|------|------|------|------| +| `/sse/` | GET | SSE 事件流订阅 | ❌ 已禁用 | +| `/sse/push/` | POST | 推送测试事件 | ✅ 可用 | +| `/sse/status/` | GET | 获取连接状态 | ✅ 可用 | +| `/sse/shutdown/` | POST | 关闭商户连接 | ✅ 可用 | + +--- + +## 3. 核心设计原则 + +### 3.1 透明转发架构（关键设计） + +**Golang SSE 服务应该是"不关心业务内容的转发层"**，这样可以避免 Django 和 Golang 之间的消息类型重复定义。 + +``` +┌─────────────┐ 任意 JSON ┌─────────────┐ 原样转发 ┌─────────────┐ +│ Django │ ─────────────────► │ Golang SSE │ ─────────────────► │ 前端 │ +│ (业务层) │ merchant_id + │ (转发层) │ │ (消费者) │ +└─────────────┘ payload └─────────────┘ └─────────────┘ +``` + +**核心原则**： + +1. **Django 是唯一定义消息类型的地方** - 所有业务事件类型只在 Django 定义 +2. **Golang 只负责三件事** - 认证、路由（按 merchant_id）、转发 +3. **Golang 不解析业务内容** - 使用 `json.RawMessage` 原样转发 +4. **新增消息类型 = 只改 Django** - Golang 服务一旦写好几乎不需要再改 + +**Golang 内部推送接口**： + +```go +// 只需要这一个结构体，永远不变 +type BroadcastRequest struct { + MerchantID int64 `json:"merchant_id"` + Payload json.RawMessage `json:"payload"` // 不解析，直接转发 +} + +func handleBroadcast(w http.ResponseWriter, r *http.Request) { + var req BroadcastRequest + json.NewDecoder(r.Body).Decode(&req) + + // 直接转发给该商户的所有连接，不解析 payload + broadcastToMerchant(req.MerchantID, req.Payload) +} +``` + +**Django 推送封装**： + +```python +# sse/client.py - Django 调用 Golang SSE 服务的客户端 + +import requests +from django.conf import settings + +SSE_SERVICE_URL = getattr(settings, 'SSE_SERVICE_URL', 'http://localhost:8080') + +def push_to_sse(merchant_id: int, payload: dict): + """向 Golang SSE 服务推送消息（透明转发）""" + try: + requests.post( + f"{SSE_SERVICE_URL}/internal/broadcast", + json={ + "merchant_id": merchant_id, + "payload": payload # 任意 JSON，Golang 不解析 + }, + timeout=5 + ) + except Exception as e: + logger.warning(f"SSE push failed: {e}") +``` + +### 3.2 多租户隔离 + +系统是多商户（Multi-tenant）架构，SSE 消息必须按商户隔离： + +- 每个用户通过 JWT Token 中的 `user.employee.merchant_id` 确定所属商户 +- 消息只推送给同一商户下的所有连接客户端 +- 不同商户之间的消息完全隔离 + +### 3.3 认证机制 + +使用 JWT（JSON Web Token）认证： + +``` +Authorization: Bearer +``` + +认证流程： +1. 从请求头提取 Bearer Token +2. 使用 Django Simple JWT 验证 Token +3. 获取用户信息：`user = jwt_authenticator.get_user(validated_token)` +4. 获取商户 ID：`merchant_id = user.employee.merchant_id` +5. 无商户关联的用户拒绝连接 + +**重要**：Golang 服务需要能够验证 Django 签发的 JWT Token，需要共享相同的 SECRET_KEY 或使用 RS256 算法。 + +--- + +## 4. 消息规范 + +### 4.1 消息格式 + +所有 SSE 消息均为 JSON 格式，通过 `data:` 前缀发送： + +``` +data: {"type": "xxx", ...}\n\n +``` + +### 4.2 系统消息类型 + +#### 连接成功消息 + +当客户端成功建立 SSE 连接后，立即发送： + +```json +{ + "type": "connected", + "message": "SSE connection established", + "merchant_id": 1 +} +``` + +#### 心跳消息 + +保持连接活跃，使用 SSE 注释格式： + +``` +: heartbeat\n\n +``` + +或带时间戳的格式： + +``` +: heartbeat 1704067200.123\n\n +``` + +心跳间隔建议：**30 秒** + +#### 服务器关闭消息 + +服务器主动关闭连接时发送： + +```json +{ + "type": "server_shutdown", + "message": "Server shutting down your connections, please reconnect later" +} +``` + +### 4.3 业务消息类型 + +业务消息统一格式： + +```json +{ + "mode": "simple_message", + "type": "", + "message": "", + "object_id": , + "merchant_id": +} +``` + +字段说明： +- `mode`: 固定为 `"simple_message"`，标识这是业务消息 +- `type`: 事件类型，用于前端区分处理逻辑 +- `message`: 人类可读的消息内容（中文） +- `object_id`: 关联业务对象的 ID（整数或字符串） +- `merchant_id`: 商户 ID，便于客户端二次验证 + +#### 已定义的业务事件类型 + +| 事件类型 | 说明 | 触发场景 | +|----------|------|----------| +| `order_paid` | 订单已支付 | 订单支付完成 | +| `stock_change` | 库存变动 | 库存入库/出库完成 | + +**注意**：未来可能增加更多事件类型，Golang 服务应设计为可扩展的。 + +--- + +## 5. 服务接口设计 + +### 5.1 SSE 连接端点 + +**端点**: `GET /sse/` + +**请求头**: +``` +Authorization: Bearer +Accept: text/event-stream +Cache-Control: no-cache +``` + +**响应头**: +``` +Content-Type: text/event-stream +Cache-Control: no-cache +X-Accel-Buffering: no +Connection: keep-alive +``` + +**CORS 支持**: +``` +Access-Control-Allow-Origin: +Access-Control-Allow-Credentials: true +Access-Control-Allow-Methods: GET, OPTIONS +Access-Control-Allow-Headers: authorization, content-type, cache-control, accept +``` + +### 5.2 消息推送接口（透明转发） + +Django 后端通过 HTTP API 将消息推送给 Golang SSE 服务。 + +**关键设计**：Golang 不解析 `payload` 内容，只负责路由和转发。 + +**端点**: `POST /internal/broadcast` + +``` +POST /internal/broadcast +Content-Type: application/json + +{ + "merchant_id": 1, + "payload": { + "action": "invalidate", + "resources": ["stock-records"], + "message": "库存已更新", + "toast": true + } +} +``` + +**Golang 处理逻辑**： + +```go +type BroadcastRequest struct { + MerchantID int64 `json:"merchant_id"` + Payload json.RawMessage `json:"payload"` // 原样转发，不解析 +} + +func handleBroadcast(w http.ResponseWriter, r *http.Request) { + var req BroadcastRequest + json.NewDecoder(r.Body).Decode(&req) + + // 直接转发，不解析 payload 内容 + broadcastToMerchant(req.MerchantID, req.Payload) + + w.WriteHeader(http.StatusOK) +} +``` + +**备选方案：Redis Pub/Sub** + +如果需要更高的可靠性或解耦： +- Django 发布消息到 Redis Channel：`sse:merchant:{merchant_id}` +- Golang 服务订阅对应 Channel 并转发给客户端 +- 优点：异步、解耦、可靠性更高 +- 缺点：增加 Redis 依赖和复杂度 + +### 5.3 状态查询接口 + +**端点**: `GET /sse/status/` + +**响应**: +```json +{ + "status": "running", + "total_clients": 10, + "merchant_clients": 3, + "merchant_id": 1, + "message": "SSE server is running with 10 total connections, 3 for your merchant" +} +``` + +### 5.4 连接关闭接口 + +**端点**: `POST /sse/shutdown/` + +关闭指定商户的所有 SSE 连接。 + +**响应**: +```json +{ + "status": "ok", + "message": "Your merchant's SSE connections have been closed", + "merchant_id": 1, + "clients": 3 +} +``` + +--- + +## 6. 连接管理 + +### 6.1 连接存储结构 + +按商户组织连接： + +```go +// 伪代码 +type ConnectionManager struct { + // merchant_id -> []Connection + connections map[int64][]*Connection + mutex sync.RWMutex +} +``` + +### 6.2 连接生命周期 + +1. **建立连接** + - 验证 JWT Token + - 提取 merchant_id + - 注册到连接管理器 + - 发送 `connected` 消息 + +2. **维护连接** + - 定期发送心跳（30 秒） + - 监听消息队列，转发给客户端 + - 处理客户端断开 + +3. **关闭连接** + - 发送 `server_shutdown` 消息 + - 从连接管理器移除 + - 释放资源 + +### 6.3 异常处理 + +- **客户端断开**：自动检测并清理 +- **消息队列满**：丢弃旧消息或断开慢客户端 +- **认证失败**：返回 403 Forbidden +- **服务关闭**：优雅地关闭所有连接 + +--- + +## 7. 前端客户端参考 + +### 7.1 使用 Fetch API（推荐） + +```javascript +function createSSEConnection(url, token, onMessage, onError, onClose) { + const controller = new AbortController(); + + fetch(url, { + method: 'GET', + headers: { + 'Authorization': `Bearer ${token}`, + 'Accept': 'text/event-stream', + 'Cache-Control': 'no-cache', + }, + signal: controller.signal + }) + .then(response => { + if (!response.ok) throw new Error(`HTTP ${response.status}`); + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + function processBuffer() { + const lines = buffer.split('\n'); + buffer = lines.pop(); + + for (const line of lines) { + if (line.startsWith('data: ')) { + const data = line.substring(6); + try { + onMessage(JSON.parse(data)); + } catch (e) { + console.error('Parse error:', e); + } + } + } + } + + function read() { + return reader.read().then(({ done, value }) => { + if (done) { onClose?.(); return; } + buffer += decoder.decode(value, { stream: true }); + processBuffer(); + return read(); + }); + } + + return read(); + }) + .catch(onError); + + return { close: () => controller.abort() }; +} +``` + +### 7.2 事件处理示例 + +```javascript +const sse = createSSEConnection('/sse/', jwtToken, + (event) => { + switch (event.type) { + case 'connected': + console.log(`连接成功，商户: ${event.merchant_id}`); + break; + case 'stock_change': + // 刷新库存列表 + refreshStockList(); + break; + case 'order_paid': + // 显示订单支付通知 + showNotification(event.message); + break; + case 'server_shutdown': + // 准备重连 + scheduleReconnect(); + break; + } + }, + (error) => { + console.error('SSE Error:', error); + scheduleReconnect(); + }, + () => console.log('SSE Closed') +); +``` + +--- + +## 8. 前端消息处理简化方案 + +### 8.1 问题分析 + +如果每种业务事件都需要前端单独处理，会造成： +- 前端需要维护一份事件类型清单 +- 新增事件类型需要同时改 Django 和前端 +- switch-case 代码越来越长 + +### 8.2 混合消息模式（推荐） + +实际业务中存在两种不同的通知需求，需要用不同的模式处理： + +| 模式 | 用途 | 典型场景 | +|------|------|----------| +| **资源失效** | 数据刷新，无特定对象 | 库存变动后刷新列表 | +| **对象事件** | 指向特定对象的通知 | 员工被指派处理特定任务 | + +### 8.3 统一消息格式 + +```typescript +interface SSEMessage { + // 必填：动作类型 + action: 'invalidate' | 'notify' | 'connected' | 'server_shutdown'; + + // 资源失效相关（可选） + resources?: string[]; // 需要刷新的资源列表 + + // 对象事件相关（可选） + object_type?: string; // 对象类型，如 'printing-job' + object_id?: number | string; // 具体对象ID + + // 用户定向（可选） + target_user_ids?: number[]; // 目标用户ID列表，为空表示广播给所有人 + + // 通知展示（可选） + message?: string; // 用户可见的提示消息 + toast?: boolean; // 是否显示 Toast 通知 + url?: string; // 点击通知后跳转的URL +} +``` + +### 8.4 消息示例 + +**模式1：纯资源刷新**（不需要知道具体对象） + +```json +{ + "action": "invalidate", + "resources": ["stock-records", "inventory"], + "message": "库存已更新", + "toast": true +} +``` + +**模式2：对象事件**（需要指向特定对象） + +```json +{ + "action": "notify", + "object_type": "printing-job", + "object_id": 123, + "target_user_ids": [5], + "message": "您有新的印染任务 #123", + "toast": true, + "url": "/printing-jobs/123" +} +``` + +**模式3：混合**（既刷新数据，又指向特定对象） + +```json +{ + "action": "notify", + "object_type": "printing-job", + "object_id": 123, + "resources": ["printing-jobs"], + "message": "印染任务 #123 已更新", + "toast": true +} +``` + +### 8.5 目标用户过滤策略 + +**问题**：当消息需要通知特定用户时，过滤应该在哪里做？ + +| 方案 | 描述 | 优缺点 | +|------|------|--------| +| A. Django 多次推送 | Django 为每个目标用户单独调用推送 | 复杂，需要跟踪用户连接状态 | +| B. **前端过滤（推荐）** | 消息带 `target_user_ids`，前端自己过滤 | ✅ 简单，保持透明转发 | +| C. Golang 过滤 | Golang 解析 `target_user_ids` 只推给特定连接 | 打破透明转发原则 | + +**选择方案 B 的理由**： + +1. **保持透明转发** - Golang 无需改动 +2. **前端过滤逻辑极简** - 一行代码 +3. **带宽浪费可忽略** - 同商户内并发用户数有限 + +### 8.6 前端通用处理器 + +```typescript +// hooks/useSSE.ts +import { useQueryClient } from '@tanstack/react-query'; +import { message, Modal } from 'antd'; +import { useNavigate } from 'react-router-dom'; +import { useCurrentUser } from './useCurrentUser'; + +interface SSEMessage { + action: 'invalidate' | 'notify' | 'connected' | 'server_shutdown'; + resources?: string[]; + object_type?: string; + object_id?: number | string; + target_user_ids?: number[]; + message?: string; + toast?: boolean; + url?: string; +} + +export function useSSEHandler() { + const queryClient = useQueryClient(); + const navigate = useNavigate(); + const { user } = useCurrentUser(); + + const handleMessage = (event: SSEMessage) => { + // ========== 用户过滤 ========== + // 如果指定了目标用户，且当前用户不在列表中，则忽略 + if (event.target_user_ids && event.target_user_ids.length > 0) { + if (!event.target_user_ids.includes(user.id)) { + return; // 不是发给我的，忽略 + } + } + + // ========== 系统消息 ========== + if (event.action === 'connected') { + console.log('SSE 连接成功'); + return; + } + + if (event.action === 'server_shutdown') { + scheduleReconnect(); + return; + } + + // ========== 资源失效处理 ========== + if (event.resources && event.resources.length > 0) { + for (const resource of event.resources) { + queryClient.invalidateQueries({ queryKey: [resource] }); + } + } + + // ========== 通知展示 ========== + if (event.toast && event.message) { + if (event.url) { + // 可点击的通知 + message.info( + navigate(event.url!)} style={{ cursor: 'pointer' }}> + {event.message} + + ); + } else { + message.info(event.message); + } + } + }; + + return { handleMessage }; +} +``` + +### 8.7 Django 侧封装 + +```python +# sse/events.py - 统一的事件发送接口 + +from sse.client import push_to_sse + +def notify_resource_changed( + merchant_id: int, + resources: list[str], + message: str = None, + toast: bool = False +): + """ + 通知前端资源已变化，需要刷新（纯资源失效模式） + """ + push_to_sse(merchant_id, { + "action": "invalidate", + "resources": resources, + "message": message, + "toast": toast + }) + + +def notify_object_event( + merchant_id: int, + object_type: str, + object_id: int, + message: str, + target_user_ids: list[int] = None, + resources: list[str] = None, + url: str = None, + toast: bool = True +): + """ + 通知特定对象的事件（对象事件模式） + + Args: + merchant_id: 商户ID + object_type: 对象类型，如 'printing-job', 'plate-order' + object_id: 对象ID + message: 用户可见的消息 + target_user_ids: 目标用户ID列表，为空表示通知所有人 + resources: 可选，同时触发哪些资源刷新 + url: 可选，点击通知后跳转的URL + toast: 是否显示 Toast + """ + payload = { + "action": "notify", + "object_type": object_type, + "object_id": object_id, + "message": message, + "toast": toast + } + + if target_user_ids: + payload["target_user_ids"] = target_user_ids + if resources: + payload["resources"] = resources + if url: + payload["url"] = url + + push_to_sse(merchant_id, payload) + + +# ========== 业务代码调用示例 ========== + +# 示例1：库存变动（纯资源刷新，通知所有人） +def on_stock_change(merchant_id, record_id): + notify_resource_changed( + merchant_id=merchant_id, + resources=["stock-records", "inventory"], # 资源为空意味着是一则普通的消息通知 + message="库存已更新", + toast=True + ) + +# 示例2：任务指派（对象事件，只通知被指派的人） +def on_job_assigned(merchant_id, job_id, assigned_user_id): + notify_object_event( + merchant_id=merchant_id, + object_type="printing-job", + object_id=job_id, + message=f"您有新的印染任务 #{job_id}", + target_user_ids=[assigned_user_id], # 可能为空意味着需要全员接收该消息 + resources=["printing-jobs"], # 同时刷新任务列表 + url=f"/printing-jobs/{job_id}" + ) + +# 示例3：订单状态变更（通知相关人员） +def on_order_status_changed(merchant_id, order_id, related_user_ids): + notify_object_event( + merchant_id=merchant_id, + object_type="printing-order", + object_id=order_id, + message=f"订单 #{order_id} 状态已更新", + target_user_ids=related_user_ids, + resources=["printing-orders"] + ) +``` + +### 8.8 资源命名约定 + +建议 `resources` 和 `object_type` 使用与 API 路径一致的命名： + +| 名称 | 对应 API | React Query Key | +|------|----------|-----------------| +| `printing-orders` | `/api/v1/printing-orders/` | `['printing-orders']` | +| `printing-jobs` | `/api/v1/printing-jobs/` | `['printing-jobs']` | +| `plate-orders` | `/api/v1/plate-orders/` | `['plate-orders']` | +| `stock-records` | `/api/v1/stock-change-records/` | `['stock-records']` | +| `inventory` | `/api/v1/inventory/` | `['inventory']` | + +### 8.9 方案优势总结 + +| 维度 | 效果 | +|------|------| +| **新增消息类型** | 只改 Django，Golang 和前端通用处理器不变 | +| **目标用户过滤** | 前端一行代码，保持透明转发 | +| **对象导航** | 支持 `object_type` + `object_id` + `url` | +| **灵活性** | 纯刷新、对象事件、混合模式都支持 | + +--- + +## 9. 部署架构 + +### 9.1 推荐架构 + +``` + ┌─────────────────┐ + │ Nginx/CDN │ + └────────┬────────┘ + │ + ┌────────────────┼────────────────┐ + │ │ │ + ▼ ▼ ▼ + ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ + │ Django API │ │ Golang SSE │ │ Static Files │ + │ (Gunicorn) │ │ Service │ │ (CDN) │ + └───────┬───────┘ └───────┬───────┘ └───────────────┘ + │ │ + │ HTTP/Redis │ + └────────┬────────┘ + │ + ┌────────▼────────┐ + │ PostgreSQL/ │ + │ Redis │ + └─────────────────┘ +``` + +### 9.2 Nginx 配置要点 + +```nginx +location /sse/ { + proxy_pass http://golang-sse-service:8080; + proxy_http_version 1.1; + proxy_set_header Connection ''; + proxy_buffering off; + proxy_cache off; + + # SSE 超时配置 + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; + + # 禁用缓冲 + add_header X-Accel-Buffering no; +} +``` + +### 9.3 Docker Compose 示例 + +```yaml +services: + sse: + image: flower-sse:latest + ports: + - "8080:8080" + environment: + - JWT_SECRET=${SECRET_KEY} + - REDIS_URL=redis://redis:6379/0 + depends_on: + - redis +``` + +--- + +## 10. 安全考虑 + +1. **JWT 验证**：必须验证 Token 有效性和过期时间 +2. **商户隔离**：严格检查 merchant_id，防止跨商户数据泄漏 +3. **内部 API 保护**：消息推送 API 只允许内网访问 +4. **连接限制**：限制单用户/单商户的最大连接数 +5. **速率限制**：防止连接风暴 + +--- + +## 11. 监控指标 + +建议监控以下指标： + +- 总连接数 +- 各商户连接数 +- 消息推送延迟 +- 消息推送成功率 +- 连接存活时长分布 +- 心跳超时断开数 + +--- + +## 12. 迁移计划 + +### Phase 1: 开发与测试 +1. 开发 Golang SSE 服务 +2. 实现 JWT 认证（与 Django 共享密钥） +3. 实现消息推送接口 +4. 单元测试和压力测试 + +### Phase 2: 并行运行 +1. 部署 Golang SSE 服务 +2. Django 同时向新旧 SSE 推送消息 +3. 前端切换到新 SSE 端点 +4. 监控对比 + +### Phase 3: 完全迁移 +1. 废弃 Django SSE 模块 +2. 移除相关代码 +3. 更新文档 + +--- + +## 13. 附录：当前代码参考 + +### 13.1 核心服务函数签名 + +```python +# sse/services.py + +def push_connection(merchant_id, conn_queue): + """将新连接添加到指定商户的连接集合""" + +def remove_connection(merchant_id, conn_queue): + """从指定商户移除连接""" + +def push_sse_event_to_merchant(merchant_id, event_data: dict): + """向指定商户的所有客户端广播事件""" + +def push_simple_message_with_object_id_to_merchant( + merchant_id, event_type: str, message: str, object_id +): + """向指定商户广播简单业务消息""" + +def shutdown_merchant_connections(merchant_id): + """关闭指定商户的所有连接""" +``` + +### 13.2 Django 中的调用示例 + +```python +# stock/services.py +from sse.services import push_simple_message_with_object_id + +# 库存变动后推送通知 +push_simple_message_with_object_id( + event_type='stock_change', + message=f'产品ID {detail.product_id} 位于 {warehouse_id} 的库存已更新', + object_id=inventory_record.id +) +``` + +---