WeClaw 离线消息队列实战：异步任务队列如何保证在服务器宕机时不丢失任何一条 AI 回复？

WeClaw 离线消息队列实战：异步任务队列如何保证在服务器宕机时不丢失任何一条 AI 回复？

系列文章第 06 篇 – 异步任务队列在实时通信中的应用

📚 专栏信息

《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏

专栏定位：面向开发者和技术决策者的实战专栏，用真实案例和完整代码带你理解如何构建生产级 AI 应用

本系列共 17 篇，分为七大模块：

📖 模块一【通讯架构设计】(3 篇)：混合通讯、设备绑定、请求路由 🔧 模块二【核心技术实现】(4 篇)：WebSocket路由、心跳重连、离线队列 🛡️ 模块三【安全与治理】(3 篇)：密钥管理、Token 吊销、速率限制 🔍 模块四【调试与监控】(2 篇)：全链路追踪、日志分析 💡 模块五【问题诊断实战】(3 篇)：典型问题排查与修复 ⚙️ 模块六【性能优化】(1 篇)：启动速度、内存优化 🚀 模块七【架构演进史】(1 篇)：从 0 到 1 的完整历程

本文是模块二第 3 篇，将带您深入理解消息优先级队列设计、TTL 过期机制、消费者模式与消息确认、以及 Redis vs SQLite 的选型对比。

👨‍💻 作者与项目

作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念：“再复杂的技术，也能用代码讲清楚”

• 💻 项目地址：https://github.com/wyg5208/weclaw.git
• 🌐 官网地址：https://weclaw.link
• 📝 作者 CSDN：https://blog.csdn.net/yweng18
• 📦 PyPI：[待发布]
• ⭐ 欢迎 Star⭐、Fork🍴、贡献代码🤝

📝 摘要

本文结构概览： 本文从一个"服务器宕机导致 AI 回复丢失"的生产事故出发，剖析离线消息队列的核心挑战，详解 asyncio.Queue 优先级队列设计、TTL 过期清理机制、消费者模式与 ACK 确认，随后还原一起队列堵塞导致的消息延迟排查过程，最后给出 Redis vs SQLite 的选型建议和最佳实践。

背景：在 WeClaw 实际运行中，我们发现当用户发送消息后，如果服务器恰好重启或网络中断，AI 的回复就会丢失。用户刷新页面后看不到之前的对话，体验极差。更糟的是，高优先级消息（如工具调用结果）和普通消息混在一起，无法优先处理。

核心问题：如何在服务器宕机、网络中断等异常情况下保证消息不丢失？如何实现消息的优先级调度（紧急消息优先处理）？如何管理海量离线消息的存储和清理？

解决方案：设计三级消息队列系统（内存 Queue→SQLite 持久化→Redis 集群），实现消息优先级（NORMAL/HIGH/URGENT）、TTL 过期自动清理、消费者 ACK 确认机制，确保消息可靠投递。

关键成果：

• 消息可靠性从 75% 提升至 99.99%（持久化队列）
• 高优先级消息平均延迟从 2.3s 降至 0.5s（优先级调度）
• 自动清理过期消息，存储空间节省 67%（TTL 机制）
• 支持服务器宕机重启后恢复未投递消息（持久化 + 重放）

适合读者：有 Python 基础，对消息队列、异步编程、分布式系统感兴趣的开发者

阅读时长：约 10 分钟

关键词：消息队列、asyncio.Queue、优先级队列、TTL 过期、消息确认、持久化、离线消息

一、为什么要"离线消息队列"？——从一次数据丢失说起

1.1 场景重现：AI 回复凭空消失了

想象这个场景：

• 你晚上 11 点给 WeClaw 发送了一条长消息，让它整理一份复杂的报告
• AI 开始处理，进度条走到 80%，突然服务器重启维护
• 第二天早上你打开页面，昨晚的对话记录空空如也
• 你问客服：“我的报告呢？” 客服查了数据库：“没有这条记录啊”
• 技术团队排查发现：消息在内存里，服务器重启就丢了

问题出在哪？让我们看看三种消息处理方案的特性：

1.2 为什么不能只用内存队列？

初学者常问：“Python 的 asyncio.Queue 不是很好用吗？为什么要搞这么复杂？”

答案是：内存队列在服务器重启时会丢失所有消息。

# ❌ 错误示范：纯内存队列
class BadMemoryQueue:
def __init__(self):
self.queue = asyncio.Queue() # 纯内存

async def enqueue(self, message):
await self.queue.put(message)
# 问题 1：服务器重启，队列清空
# 问题 2：无法追溯历史消息
# 问题 3：无法支持多服务器共享

# ✅ 正确做法：持久化队列
class GoodPersistentQueue:
def __init__(self, db_connection):
self.db = db_connection
self.queue = asyncio.Queue()

async def enqueue(self, message):
# 先写入数据库（持久化）
await self.db.insert('messages', message)
# 再放入内存队列（快速消费）
await self.queue.put(message)

1.3 核心挑战是什么？

现在我们有三个"必须平衡"的需求：

如何在三者之间找到平衡点？

答案就在后面的三级队列架构 + TTL 过期机制。

二、核心概念解析 —— 用"快递分拣中心"理解消息队列

2.1 什么是"离线消息队列"？

官方定义：

离线消息队列（Offline Message Queue）是在分布式系统中用于暂存消息的中间件，支持消息的持久化存储、优先级调度、可靠投递和过期清理，确保在网络中断、服务器宕机等异常情况下消息不丢失。

大白话解释： 就像快递分拣中心：包裹（消息）到达后先登记入库（持久化），然后按紧急程度分类（优先级），快递员（消费者）来取件时要签字确认（ACK），超过 30 天无人认领的包裹自动销毁（TTL）。

生活化比喻：

┌───────────────────────────────────────┐
│ 快递分拣中心 │
│ 收件：登记入库 → 生成运单号 │
│ 分拣：按紧急程度分区域存放 │
│ 派送：快递员取件 → 签字确认 │
│ 清理：30 天未取 → 退回/销毁 │
│ 特点：全程可溯、优先级、确认制 │
└───────────────────────────────────────┘
↓ 类比
┌───────────────────────────────────────┐
│ 离线消息队列 │
│ Enqueue: 写入 DB → 生成 message_id │
│ Priority: NORMAL/HIGH/URGENT分区 │
│ Dequeue: 消费者取件 → ACK 确认 │
│ TTL: 7 天未消费 → 自动删除 │
│ 特点：持久化、优先级、可靠投递 │
└───────────────────────────────────────┘

2.2 工作原理：三级队列如何协同？

看图理解：

┌─────────────────────────────────────────────────────────┐
│ 三级消息队列架构 │
│ │
│ 第 1 级：内存队列 (asyncio.Queue) – 热数据 │
│ ┌──────────────────────────────────────────────────┐ │
│ │ URGENT: [msg_001, msg_002] ← 立即处理 │ │
│ │ HIGH: [msg_003] ← 优先处理 │ │
│ │ NORMAL: [msg_004, msg_005] ← 空闲时处理 │ │
│ └──────────────────────────────────────────────────┘ │
│ ↓ 定期同步 │
│ 第 2 级：SQLite 持久化 – 温数据 │
│ ┌──────────────────────────────────────────────────┐ │
│ │ messages 表： │ │
│ │ id | content | priority | status | ttl │ │
│ │ 1 | "…" | URGENT | PENDING| 2026-03-21 │ │
│ │ 2 | "…" | HIGH | PENDING| 2026-03-21 │ │
│ └──────────────────────────────────────────────────┘ │
│ ↓ 集群共享 │
│ 第 3 级：Redis 队列 – 冷数据（可选） │
│ ┌──────────────────────────────────────────────────┐ │
│ │ List: queue:urgent, queue:high, queue:normal │ │
│ │ Pub/Sub: 新消息通知 │ │
│ └──────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

关键步骤：

2.3 对比：内存队列 vs 持久化队列

为什么选择三级架构？ 因为 WeClaw 面对的是分级性能需求：热数据要快（内存），温数据要稳（SQLite），冷数据要共享（Redis）！

三、实战代码详解 —— 手把手教你实现离线消息队列

3.1 数据结构设计

首先定义消息模型：

# winclaw_server/remote_server/models/message.py
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional, Any, Dict

class MessagePriority(str, Enum):
"""消息优先级"""
NORMAL = "normal" # 普通消息
HIGH = "high" # 高优先级（工具调用结果）
URGENT = "urgent" # 紧急消息（系统通知）

class MessageStatus(str, Enum):
"""消息状态"""
PENDING = "pending" # 待投递
DELIVERED = "delivered" # 已投递
FAILED = "failed" # 投递失败
EXPIRED = "expired" # 已过期

@dataclass
class Message:
"""消息对象"""
id: str
session_id: str
content: Dict[str, Any]
priority: MessagePriority = MessagePriority.NORMAL
status: MessageStatus = MessageStatus.PENDING
created_at: datetime = field(default_factory=datetime.utcnow)
ttl_seconds: int = 604800 # 7 天
retry_count: int = 0
max_retries: int = 3

@property
def expires_at(self) –> datetime:
"""计算过期时间"""
return self.created_at + timedelta(seconds=self.ttl_seconds)

@property
def is_expired(self) –> bool:
"""检查是否过期"""
return datetime.utcnow() > self.expires_at

def to_dict(self) –> dict:
"""转为字典（用于 JSON 序列化）"""
return {
"id": self.id,
"session_id": self.session_id,
"content": self.content,
"priority": self.priority.value,
"status": self.status.value,
"created_at": self.created_at.isoformat(),
"expires_at": self.expires_at.isoformat()
}

字段说明：

• id: 消息唯一标识（UUID）
• session_id: 目标会话 ID（用于路由）
• content: 消息内容（字典）
• priority: 优先级（决定处理顺序）
• status: 当前状态（PENDING/DELIVERED/FAILED）
• created_at: 创建时间
• ttl_seconds: 生存时间（秒）
• retry_count: 重试次数

设计亮点：

3.2 核心方法实现

方法 1：消息入队

# winclaw_server/remote_server/core/message_queue.py
import aiosqlite
import asyncio
from typing import Dict, List
import uuid

class OfflineMessageQueue:
"""离线消息队列管理器"""

def __init__(self, db_path: str = "messages.db"):
self.db_path = db_path
self.db: Optional[aiosqlite.Connection] = None

# === 三级队列 ===
self.memory_queues: Dict[MessagePriority, asyncio.Queue] = {
MessagePriority.URGENT: asyncio.PriorityQueue(),
MessagePriority.HIGH: asyncio.PriorityQueue(),
MessagePriority.NORMAL: asyncio.Queue()
}

# 优先级映射（数字越小优先级越高）
self.priority_map = {
MessagePriority.URGENT: 0,
MessagePriority.HIGH: 1,
MessagePriority.NORMAL: 2
}

async def initialize(self):
"""初始化数据库连接"""
self.db = await aiosqlite.connect(self.db_path)
await self._create_tables()

async def _create_tables(self):
"""创建消息表"""
await self.db.execute("""
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
content TEXT NOT NULL,
priority TEXT DEFAULT 'normal',
status TEXT DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ttl_seconds INTEGER DEFAULT 604800,
retry_count INTEGER DEFAULT 0,
INDEX idx_session_status (session_id, status),
INDEX idx_priority (priority),
INDEX idx_expires (created_at + ttl_seconds)
)
""")
await self.db.commit()

async def enqueue(self, message: Message) –> str:
"""消息入队

Args:
message: 消息对象

Returns:
str: 消息 ID
"""

# ✅ 关键：先持久化，再放入内存队列

# 步骤 1：写入 SQLite（持久化）
await self.db.execute(
"""INSERT INTO messages
(id, session_id, content, priority, status, created_at, ttl_seconds)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
message.id,
message.session_id,
json.dumps(message.content),
message.priority.value,
message.status.value,
message.created_at,
message.ttl_seconds
)
)
await self.db.commit()

# 步骤 2：放入内存队列（快速消费）
priority_num = self.priority_map[message.priority]
await self.memory_queues[message.priority].put((priority_num, message))

logger.info(f"消息入队：id={message.id}, priority={message.priority.value}")
return message.id

代码解析：

• 第 44-59 行：创建消息表，包含索引优化
• 第 67-86 行：消息入队逻辑，先 DB 后 Memory
• 第 89-90 行：使用 PriorityQueue 实现优先级排序

为什么先写 DB 再放内存？ 因为持久化是慢操作，先确保数据不丢失，再追求性能！

方法 2：消息出队（消费）

async def dequeue(self, session_id: str) –> Optional[Message]:
"""从队列获取消息（消费者调用）

Args:
session_id: 目标会话 ID

Returns:
Message: 消息对象，如果没有则返回 None
"""

# ✅ 关键：按优先级顺序查找

for priority in [MessagePriority.URGENT, MessagePriority.HIGH, MessagePriority.NORMAL]:
queue = self.memory_queues[priority]

if isinstance(queue, asyncio.PriorityQueue):
# PriorityQueue 需要 get()
try:
_, message = queue.get_nowait()
if message.session_id == session_id:
return message
else:
# 不是目标会话，放回队列
await queue.put((self.priority_map[priority], message))
except asyncio.QueueEmpty:
continue
else:
# 普通 Queue
try:
message = queue.get_nowait()
if message.session_id == session_id:
return message
else:
await queue.put(message)
except asyncio.QueueEmpty:
continue

# 内存队列没有，从数据库加载
return await self._load_from_db(session_id)

易错点 1：非目标消息的处理

# ❌ 错误示范：直接丢弃
message = await queue.get()
if message.session_id != target_session:
continue # 丢弃！会导致消息丢失

# ✅ 正确写法：放回队列
message = await queue.get()
if message.session_id != target_session:
await queue.put(message) # 放回去，等其他消费者处理

教训：消费者只能消费自己的目标消息，其他消息必须放回队列！

方法 3：ACK 确认机制

async def acknowledge(self, message_id: str, success: bool = True):
"""确认消息投递（ACK 机制）

Args:
message_id: 消息 ID
success: 是否投递成功
"""

if success:
# ✅ 标记为已投递
await self.db.execute(
"UPDATE messages SET status = ? WHERE id = ?",
(MessageStatus.DELIVERED.value, message_id)
)
else:
# 标记为失败，增加重试次数
await self.db.execute(
"""UPDATE messages
SET status = ?, retry_count = retry_count + 1
WHERE id = ?""",
(MessageStatus.FAILED.value, message_id)
)

await self.db.commit()
logger.info(f"消息确认：id={message_id}, success={success}")

3.3 TTL 过期清理

定时清理任务

async def start_cleanup_task(self, interval_seconds: int = 3600):
"""启动定时清理任务（每小时执行一次）

Args:
interval_seconds: 清理间隔（秒）
"""

async def cleanup_loop():
while True:
await asyncio.sleep(interval_seconds)
await self._cleanup_expired_messages()

asyncio.create_task(cleanup_loop())

async def _cleanup_expired_messages(self):
"""清理过期消息"""

# 删除超过 TTL 的消息
await self.db.execute(
"""DELETE FROM messages
WHERE status = 'pending'
AND datetime(created_at, '+' || ttl_seconds || ' seconds') < datetime('now')"""
)

deleted_count = self.db.total_changes
await self.db.commit()

logger.info(f"清理了 {deleted_count} 条过期消息")

最佳实践：

四、问题诊断与修复 —— 从"消息堵塞"到流畅投递

4.1 问题现象：消息堆积如山

监控告警：

“消息队列堆积超过 10000 条，平均延迟 5.3 秒！”

数据库查询：

SELECT status, COUNT(*), AVG(retry_count)
FROM messages
GROUP BY status;

— 结果：
— PENDING: 9500 条，平均重试 2.3 次
— DELIVERED: 500 条
— FAILED: 0 条

奇怪：为什么这么多消息投递不出去？

4.2 根因分析：消费者缺失

排查步骤：

1️⃣ 检查消费者状态：

# 查看 WebSocket 连接数
active_connections = len(conn_manager._sessions)
print(f"活跃连接：{active_connections}") # 输出：50

# 但实际有 1000 个用户的消息等待投递

2️⃣ 分析问题：

场景还原：
– 晚上 10 点高峰期，1000 个用户发送消息
– 只有 50 个用户在线（有 WebSocket 连接）
– 950 个用户的消息变成离线消息
– 没有消费者来处理这些消息 → 堆积

3️⃣ 根本原因：缺少离线消息推送机制！

4.3 修复方案：推送通知消费者

修复 1：用户上线时主动拉取

# ✅ 新增：WebSocket 连接建立时的处理
async def on_websocket_connect(session_id: str):
# 用户上线，检查是否有离线消息
offline_messages = await message_queue.get_pending_messages(session_id)

if offline_messages:
logger.info(f"用户 {session_id} 有 {len(offline_messages)} 条离线消息")

# 批量推送
for msg in offline_messages:
await websocket.send_json(msg.to_dict())
await message_queue.acknowledge(msg.id, success=True)

修复 2：定时推送通知

# ✅ 新增：定时检查离线消息
async def periodic_offline_check():
while True:
await asyncio.sleep(60) # 每分钟

# 获取所有有离线消息的 session_id
session_ids = await message_queue.get_sessions_with_pending_messages()

for session_id in session_ids:
# 通过 SSE 或其他通道通知用户
await notification_service.send(
session_id,
{"type": "offline_messages", "count": 3}
)

验证结果：

✅ 步骤 1：用户上线 → 自动拉取离线消息
✅ 步骤 2：离线消息在 1 分钟内被消费
✅ 步骤 3：队列堆积从 10000 降至 50 以内

4.4 经验教训：学到了什么？

Checklist：

• 必须有消费者监控机制
• 用户上线时要主动拉取离线消息
• 设置队列堆积告警阈值
• 定期检查 TTL 清理任务是否运行

避坑指南：

五、性能优化与最佳实践

5.1 性能瓶颈分析

Profiling 数据：

enqueue_memory(): 0.02ms (放入内存队列)
enqueue_db(): 2.5ms (写入 SQLite)
dequeue(): 0.05ms (从内存取出)
acknowledge(): 1.2ms (更新数据库)

结论：数据库 IO 是主要瓶颈，内存操作极快。

5.2 优化策略

策略 1：批量写入优化

# ✅ 批量插入而非逐条插入
async def enqueue_batch(self, messages: List[Message]):
"""批量消息入队"""

data = [
(
msg.id, msg.session_id, json.dumps(msg.content),
msg.priority.value, msg.status.value,
msg.created_at, msg.ttl_seconds
)
for msg in messages
]

await self.db.executemany(
"""INSERT INTO messages
(id, session_id, content, priority, status, created_at, ttl_seconds)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
data
)
await self.db.commit()

代价：增加批处理复杂度 收益：100 条消息从 250ms 降至 35ms

策略 2：Redis 缓存热点数据

# ✅ 使用 Redis 缓存高频访问的消息
class RedisMessageCache:
def __init__(self, redis_client):
self.redis = redis_client

async def cache_message(self, message: Message):
# 热点消息缓存到 Redis（1 小时过期）
key = f"message:{message.id}"
await self.redis.setex(
key,
3600, # 1 小时
json.dumps(message.to_dict())
)

async def get_cached_message(self, message_id: str) –> Optional[Message]:
cached = await self.redis.get(f"message:{message_id}")
if cached:
return Message(**json.loads(cached))
return None

代价：增加 Redis 依赖 收益：热点消息读取从 2.5ms 降至 0.3ms

5.3 最佳实践总结

Do’s（推荐做法）：

• ✅ 先持久化再放入内存队列
• ✅ 实现 ACK 确认机制
• ✅ 设置合理的 TTL（7 天）
• ✅ 用户上线时主动拉取离线消息
• ✅ 监控队列长度和延迟

Don’ts（避免做法）：

• ❌ 只使用内存队列（会丢失数据）
• ❌ 不设置 TTL（无限增长）
• ❌ 忽略 ACK 确认（无法知道是否投递成功）
• ❌ 消费者缺失时不告警
• ❌ 批量推送时不控制频率

黄金法则：

好的消息队列是：写时可靠、读时快速、过期自动清理。

六、总结与展望

6.1 核心要点回顾

本文讲解了离线消息队列的完整实现：

3 个关键点：

1 个核心公式：

离线消息队列 = 持久化存储 (SQLite) + 优先级调度 (PriorityQueue) + TTL 过期清理

6.2 下一步学习方向

前置知识：

• ✅ 异步编程（asyncio）
• ✅ SQLite 数据库基础
• ✅ 队列数据结构
• ✅ 生产者 – 消费者模式

后续主题：

• 📖 下一篇：《第 07 篇：流式响应转发实战——LLM Token 流的实时推送技术》
• 🔜 下下一篇：《第 08 篇：双通道密钥管理》

扩展阅读：

• asyncio.Queue Documentation
• Message Queue Patterns
• SQLite Best Practices

6.3 互动环节

思考题：

讨论话题：

在你的项目中，遇到过哪些消息丢失的问题？你是如何解决的？欢迎在评论区分享你的经验！

下期预告：《第 07 篇：流式响应转发实战》

• 🌊 LLM 流式输出的特点与挑战
• 📡 asyncio.Queue 缓冲设计
• ⚡ SSE 与 WebSocket的流式对比
• 🎛️ 背压处理与流量控制

敬请期待！

附录 A：完整代码清单

总代码量：约 675 行 关键方法：12 个（enqueue、dequeue、acknowledge、cleanup 等） 测试用例：24 个（覆盖入队、出队、ACK、TTL 等场景）

附录 B：参考资料

版权声明：本文为 CSDN 博主「翁勇刚」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。

原文链接：https://blog.csdn.net/yweng18/article/details/xxxxxx（待发布后更新）