07 - 线程安全与并发控制

学习目标

理解 CowAgent 如何在多线程环境中保证线程安全，包括消息锁、渠道并发控制、MCP 工具字典的 GIL 安全操作。

项目实践

锁策略总览

CowAgent 使用多种锁策略保护共享状态：

Agent 消息锁

messages_lock 是 Agent 中最频繁使用的锁，保护 messages 列表的所有读写操作：

# 读操作
with self.messages_lock:
    messages_copy = self.messages.copy()
    original_length = len(self.messages)

# 写操作
with self.messages_lock:
    self.messages = list(executor.messages)
    self._last_run_new_messages = list(executor.messages[trim_adjusted_start:])

# 边界清理
with self.messages_lock:
    msg_count = len(agent.messages)
    if msg_count == 0:
        get_conversation_store().clear_session(session_id)

锁的范围覆盖：

Agent.run_stream() 中读取和同步消息
AgentBridge.agent_reply() 中持久化消息
ChatService.run() 中同步 executor 消息

渠道并发控制

ChatChannel 使用 BoundedSemaphore 控制每 session 的并发：

# 每 session 初始化
self.sessions[session_id] = [
    Dequeue(),                                    # 消息队列
    threading.BoundedSemaphore(conf().get("concurrency_in_session", 1))  # 并发限制
]

# 消费循环
if semaphore.acquire(blocking=False):  # 非阻塞获取
    context = context_queue.get()
    future = handler_pool.submit(self._handle, context)
    future.add_done_callback(lambda _: semaphore.release())  # 完成后释放

BoundedSemaphore(1) 确保同一 session 的消息串行处理，不会乱序。不同 session 之间可以并行。

MCP 工具字典的 GIL 安全

_mcp_tool_instances 是一个 dict[str, McpTool]，在后台 loader 线程中写入，在多个 Agent 线程中读取：

# 写入（后台 loader 线程）
self._mcp_tool_instances[tool_name] = mcp_tool  # GIL 安全的 dict 赋值

# 读取（Agent 线程）
current = self._mcp_tool_instances
for name, tool in list(current.items()):  # list() 创建快照，避免并发迭代
    ...

Python 的 GIL 保证了简单 dict 赋值的原子性。list(current.items()) 创建迭代开始时的快照，即使迭代过程中字典被修改也不会抛出 RuntimeError。

渠道停止的优雅处理

ChannelManager.stop() 使用 pop-under-lock 模式避免死锁：

def stop(self, channel_name=None):
    # 1. 在锁下 pop，获取要停止的目标
    with self._lock:
        to_stop = [(name, self._channels.pop(name), self._threads.pop(name))]

    # 2. 在锁外执行 ch.stop() 和 th.join()，避免死锁
    for name, ch, th in to_stop:
        ch.stop()
        th.join(timeout=5)

    # 3. 超时后强制中断（仅当 ch.stop() 未成功时）
    if th.is_alive() and not graceful:
        self._interrupt_thread(th, name)  # PyThreadState_SetAsyncExc

如果在锁内调用 th.join()，被 join 的线程可能需要获取同一个锁才能完成清理——这就是死锁。

问题与规避

问题	CowAgent 的规避策略
多线程同时修改 messages 列表	`messages_lock` 保护所有读写
同一 session 消息乱序	`BoundedSemaphore(1)` 串行化
迭代 MCP 工具字典时并发修改	`list(dict.items())` 快照模式
渠道停止时死锁	pop-under-lock + 锁外 join
调度器重复初始化	模块级锁 + 双重检查
线程强制中断失败	`_interrupt_thread()` 检查 `res == 1`，失败时回滚

设计取舍

为什么不用 asyncio.Lock？

CowAgent 使用 threading.Lock 而非 asyncio.Lock，因为整个项目基于 threading 并发。asyncio.Lock 只能在 async 函数中使用，而 CowAgent 的核心代码路径都是同步的（run_stream()、_handle() 等）。

为什么 BoundedSemaphore 而非 Lock？

BoundedSemaphore 比 Lock 更灵活：

acquire(blocking=False) 支持非阻塞尝试——如果 semaphore 已被占用，直接跳过而不等待
初始值可以通过 concurrency_in_session 配置调整——未来如果想允许同一 session 并发处理（如多个用户在同一群聊），只需增大这个值
BoundedSemaphore 有上限检查，防止 release() 调用过多导致信号量膨胀

参考来源

版本信息

📅 2026-05-30 🏷️ 3dd4b84 🔀 3dd4b84