forked from erp-dev/erp
fix: added has_started into printing jobs list srz
This commit is contained in:
@@ -279,6 +279,7 @@ class PrintingJobListSerializer(serializers.ModelSerializer):
|
|||||||
product_image_url = serializers.SerializerMethodField()
|
product_image_url = serializers.SerializerMethodField()
|
||||||
status = serializers.CharField(read_only=True)
|
status = serializers.CharField(read_only=True)
|
||||||
work_state_display = serializers.SerializerMethodField()
|
work_state_display = serializers.SerializerMethodField()
|
||||||
|
has_started = serializers.BooleanField(read_only=True)
|
||||||
is_completed = serializers.BooleanField(read_only=True)
|
is_completed = serializers.BooleanField(read_only=True)
|
||||||
business_object_id = serializers.SerializerMethodField()
|
business_object_id = serializers.SerializerMethodField()
|
||||||
batch_advance_records = serializers.SerializerMethodField()
|
batch_advance_records = serializers.SerializerMethodField()
|
||||||
@@ -287,7 +288,7 @@ class PrintingJobListSerializer(serializers.ModelSerializer):
|
|||||||
model = models.PrintingJob
|
model = models.PrintingJob
|
||||||
fields = [
|
fields = [
|
||||||
'id', 'original_id', 'printing_order', 'printing_order_id', 'product', 'product_name',
|
'id', 'original_id', 'printing_order', 'printing_order_id', 'product', 'product_name',
|
||||||
'product_image_url',
|
'product_image_url', 'has_started',
|
||||||
'quantity', 'unit', 'size', 'pieces', 'description',
|
'quantity', 'unit', 'size', 'pieces', 'description',
|
||||||
'work_state', 'work_state_display',
|
'work_state', 'work_state_display',
|
||||||
'status', 'is_completed', 'business_object_id',
|
'status', 'is_completed', 'business_object_id',
|
||||||
|
|||||||
909
docs/sse_refactor.md
Normal file
909
docs/sse_refactor.md
Normal file
@@ -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 <JWT_TOKEN>
|
||||||
|
```
|
||||||
|
|
||||||
|
认证流程:
|
||||||
|
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": "<event_type>",
|
||||||
|
"message": "<human_readable_message>",
|
||||||
|
"object_id": <related_object_id>,
|
||||||
|
"merchant_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 <JWT_TOKEN>
|
||||||
|
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: <request_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(
|
||||||
|
<span onClick={() => navigate(event.url!)} style={{ cursor: 'pointer' }}>
|
||||||
|
{event.message}
|
||||||
|
</span>
|
||||||
|
);
|
||||||
|
} 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
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
Reference in New Issue
Block a user