fix(duplicate): 修复竞态条件导致的同一条消息被处理两次
问题:两条自动回复在同一时刻(11:59:22)触发,匹配度不同(118.71和14.19), 说明是竞态条件导致同一用户消息被两个线程同时处理。 根因:DB查重在无锁情况下运行,两个线程可同时通过DB查重检查, 然后都加入_processing_messages集合,导致同一条消息被处理两次。 修复:将DB查重和hash锁定合并到同一个锁内: 1. 先计算msg_hash 2. 获取锁后,先检查是否已在_processing_messages中 3. 再执行DB查重 4. 最后加入_processing_messages 这样确保同一消息的多次调用只有一个能通过所有检查。
This commit is contained in:
@@ -914,19 +914,26 @@ class MsgBotApp(ctk.CTk):
|
||||
self._log_queue.put(" ⚠ 知识库未加载,无法搜索答案")
|
||||
return
|
||||
|
||||
# ⭐ DB查重:检查用户消息是否已被回复过(防重复回复)
|
||||
if self.monitor._msg_history:
|
||||
if self.monitor._msg_history.is_user_message_replied(platform, msg, time.time()):
|
||||
self._log_queue.put(f"[{platform.upper()}] ⚠ 消息已被回复过(DB查重),跳过: {msg[:40]}")
|
||||
return
|
||||
|
||||
# ⭐ 跨平台互斥:同一消息只允许一个平台处理
|
||||
# ⭐ 计算消息hash,用于锁内去重(防竞态条件)
|
||||
import hashlib
|
||||
msg_hash = hashlib.md5(msg.strip().encode('utf-8')).hexdigest()
|
||||
|
||||
# ⭐ 在锁内完成所有检查,防止竞态条件:
|
||||
# 原来DB查重在无锁情况下运行,两个线程可同时通过DB查重,
|
||||
# 然后都加入_processing_messages,导致同一条消息被处理两次。
|
||||
with self._processing_lock:
|
||||
# 检查是否正在处理(同一消息的第二次调用)
|
||||
if msg_hash in self._processing_messages:
|
||||
self._log_queue.put(f"[{platform.upper()}] ⚠ 消息已被其他平台处理,跳过: {msg[:40]}")
|
||||
self._log_queue.put(f"[{platform.upper()}] ⚠ 消息正在处理中(锁去重),跳过: {msg[:40]}")
|
||||
return
|
||||
|
||||
# DB查重:检查用户消息是否已被回复过(防重复回复)
|
||||
if self.monitor._msg_history:
|
||||
if self.monitor._msg_history.is_user_message_replied(platform, msg, time.time()):
|
||||
self._log_queue.put(f"[{platform.upper()}] ⚠ 消息已被回复过(DB查重),跳过: {msg[:40]}")
|
||||
return
|
||||
|
||||
# 加入正在处理集合
|
||||
self._processing_messages.add(msg_hash)
|
||||
|
||||
# 在后台线程搜索
|
||||
|
||||
Reference in New Issue
Block a user