Python 同步原语与队列

锁、事件与队列协调并发任务；优先用队列传递所有权，降低共享内存复杂度。

#concept #growing #dev #python #python

[!info] related notes

所属 MOC：并发与异步 · 学习路线

前置：GIL 与线程 · 异步

Python 同步原语与队列

这个概念为什么出现

多线程/多协程共享计数器、连接池、缓冲时，需要协调。锁能保临界区；队列常常更简单：把数据所有权在任务间传递，减少共享写。

[!abstract] 一句话理解同步原语保护共享状态；队列实现生产者-消费者解耦。能队列就别到处加锁。

最小可运行示例

先把示例放进业务场景，再看代码：

场景：异步爬虫：生产者放 URL，消费者写结果

# async_queue_demo.py
# 示例意图：队列分发 URL，两个 worker 在同一循环线程追加结果；不实际抓取网页。
# 教学点：
# - asyncio.Queue；
# - worker 消费；
# - join 等待完成。

import asyncio


async def worker(name: str, q: asyncio.Queue[str], out: list[str]) -> None:
    while True:
        url = await q.get()
        try:
            await asyncio.sleep(0.05)
            out.append(f"{name}:{url}")
        finally:
            q.task_done()


async def main() -> None:
    q: asyncio.Queue[str] = asyncio.Queue(maxsize=2)
    out: list[str] = []
    workers = [asyncio.create_task(worker(f"w{i}", q, out)) for i in range(2)]
    try:
        for u in ["u1", "u2", "u3"]:
            await q.put(u)
        await q.join()
    finally:
        for w in workers:
            w.cancel()
        await asyncio.gather(*workers, return_exceptions=True)
    print(sorted(out))


if __name__ == "__main__":
    asyncio.run(main())

建议运行：

python async_queue_demo.py

一种输出（worker 分配可能不同，应验证 URL 恰好各出现一次）：

['w0:u1', 'w0:u3', 'w1:u2']

分配给不同 worker 时字符串本身也会不同，不能只按排序后的整串结果判断正确性。

结合场景再看三个关注点

Queue 自带协调，减少手写条件变量。
task_done 减少未完成计数，join 等它归零；本例 finally 即使遇到失败/取消也会确认，所以 join 不证明业务成功。示例只覆盖正常模拟路径；实际 worker 失败需监督，否则生产者或 join 可能一直等待。
线程世界对应 queue.Queue。

核心概念与准确模型

原语	场景
Lock/Mutex	临界区
RLock	可重入
Event	可 set/clear 的状态标志；不是累计事件计数器
Semaphore	限制同时持有许可数；不直接限制每秒请求数
Queue	传递工作项

边界情况与反直觉行为

锁粒度太大吞吐下降。
未 join 就退出丢任务。
asyncio 原语一般不具线程安全性；同循环任务用异步原语，线程共享状态用线程原语。threading.Lock 的阻塞获取可能堵住事件循环，RLock 又按线程重入，不能用来互斥同线程的多个 Task。
Queue 传递的是对象引用，不会强制转移所有权或复制对象；发送后仍修改同一对象会造成共享状态问题。

常见误区

[!warning] 常见误区：所有共享都靠全局 list+锁错误理解：唯一模式。
正确模型：优先所有权转移（队列）与不可变消息。

工程实践

有界队列做背压
超时获取避免永久阻塞
指标：队列深度

本节总结

同步工具保护正确性；队列简化架构。并发代码的可读性往往来自“少共享”。

自测题

为何队列能减少锁？
task_done 的作用？

参考答案

按约定转移处理责任并避免发送后继续修改，才能减少共享写；队列本身不实施对象所有权隔离。
每次成功 get 后对应调用一次，减少未完成计数；它不验证处理成功，也不负责终止 worker。

延伸阅读与资料来源

资料	类型	支撑内容
asyncio.Queue	文档	异步队列
threading	文档	线程原语