异步协程中同步原语
如果了解过多线程的的同步原语的话，我们应该知道在线程也同样的存在类似说法。首先我们的需明确一点，同步原语主要是指：在控制多个线程（或协程）的环境中，当我们涉及到访问共享资源的时候，我们需要定制某个机制用来避免竞态条件和其他并发问题- 这个就是所谓的同步原语的机制。
回顾我们了解过的多线程环境中，Python 的 threading 模块为我们提供了以下的同步原语有以下的几种：
锁（Lock）:
threading.Lock 是一个互斥锁，用于防止多个线程同时访问共享资源。 acquire() 方法用于获取锁，release() 方法用于释放锁。 条件（Condition）:
threading.Condition 基于锁，提供了 wait() 和 notify() 方法。 wait() 方法用于等待条件变量的值变为 True，notify() 方法用于唤醒一个或所有等待的线程。 信号量（Semaphore）:
threading.Semaphore 是一个计数信号量，允许一定数量的线程同时访问共享资源。 acquire() 方法用于获取信号量， release() 方法用于释放信号量。 事件（Event）:
threading.Event 是一个简单的线程同步机制，用于通知一组线程某个事件是否发生。 wait() 方法用于等待事件 set() 方法用于设置事件。 互斥锁（RLock）:
threading.RLock 是一个读写锁，提供了读操作和写操作的互斥控制。
同理，我们的异步编程中Python的asyncio 模块也提供了类似的以下几种同步原语：
事件（Event）:
asyncio.Event 用于在协程之间传递事件通知。 wait() 方法用于等待事件被设置 set() 方法用于设置事件 is_set() 方法用于检查事件是否被设置。 条件（Condition）:
asyncio.Condition 基于事件，提供了 wait() 和 notify() 方法。 wait() 方法用于等待条件变量的值变为 True notify() 方法用于唤醒一个或所有等待的协程。 notify_all() 方法用于多个协程等待同一个条件的情况下，当条件满足时，通知所有等待的协程继续执行。 locked()方法用于检查 condition 是否被锁定。 acquire()方法用于尝试获取condition 锁。 release()方法用于释放 condition 锁。 信号量（Semaphore）:
asyncio.Semaphore 是一个用于控制对共享资源的访问数量的同步原语。它允许一定数量的协程同时访问共享资源。（前面我们演示过一次） 互斥锁（Lock）:
asyncio.Lock 是一个用于协程之间的互斥控制的同步原语。它允许单个协程在同一时间访问共享资源。
从字面上，我们大体可以了解到他们的用途是基本类似的，他们呢都用于控制对共享资源的访问，以避免竞态条件和提高程序的准确性，且两者都是于互斥访问共享资源的基本同步机制。非要说不同的话，则主要是体现在下面几点了：
首先并发模式的不同，一个是多线程（共享内存的并发模型），一个单线程内（基于事件循环的异步并发模型） 再次就是异步的相对同步来说单线程内的异步同步原语通常更加轻量级，因为它们不需要处理线程上下文切换和共享内存的同步问题。
接下来我们详细针对上面几个同步原语的几个细节点展开。
asyncio.Event
首先我们的asyncio.Event事件主要是用于在协程之间传递事件通知。它允许一个协程通知其他协程某个事件已经发生。
详细的来说就是：当一个协程需要等待某个事件发生时，它会调用 Event 对象的 wait() 方法，这将导致协程进入等待状态，直到事件被设置，也就是其他的协程如果调用 set() 方法来设置事件，那么就会唤醒所有等待该事件的协程。
基本用法
Event(): 创建一个新的 Event 对象。 wait(): 阻塞直到事件被设置。如果事件已经被设置，wait() 将立即返回。 set(): 设置事件，唤醒所有等待 wait() 的协程。 clear(): 清除事件，使所有等待 wait() 的协程再次等待。 is_set(): 返回一个布尔值，表示事件是否被设置。
如以下的简单的示例代码所示：
import asyncio
async def worker1(event):
print("工作者1开始----准备处理.")
await event.wait()
print("工作者1结束----处理完成")
async def worker2(event):
print("工作者2开始----准备处理.")
await event.wait()
print("工作者2结束----处理完成")
async def set_event(event):
print("设置事件协程开始----准备设置事件.")
await asyncio.sleep(5)
event.set()
print("设置事件协程结束----事件已设置.")
async def main():
# 创建一个Event对象
event = asyncio.Event()
# 创建两个协程，一个等待事件，一个设置事件
await asyncio.gather(worker1(event),worker2(event), set_event(event))
# 运行事件循环
asyncio.run(main())
在上面的代码中，相关说明在有注释说明，这里就不再继续详细的展开，智商上面的代码，我们会开始看到：任务的开始的提示，间隔了5秒后，又同时的看到输出任务结束的提示，如下输出的结果：
工作者1开始----准备处理.
工作者2开始----准备处理.
设置事件协程开始----准备设置事件.
设置事件协程结束----事件已设置.
工作者1结束----处理完成
工作者2结束----处理完成
也就是说当我们的另一个协程进行event.set()的调用后把阻塞着的任务都给幻唤醒了。基于上面的示例结果，我们可以大致了解到它可以应用的场景有哪些：
信号通知: 当一个协程需要通知其他协程某个条件已经满足时，此时我们可以使用 Event通过设置 Event 来通知其他协程。 同步启动: 在多个协程需要同时开始执行的情况下，可以使用 Event 来同步它们的启动。比如一开始我们所有协程在 Event 上等待，直到某个条件满足，然后同时开始执行。 终止信号: 当需要优雅地关闭多个协程时，可以设置一个 Event 作为终止信号。当需要关闭程序时，设置这个事件，所有协程检查这个事件并在适当的时候退出。 心跳检测: 在需要定期检查某个条件是否满足的情况下，可以使用 Event 作为心跳机制。定期设置和清除 Event，协程在每次心跳周期检查 Event 的状态。
asyncio.Condition
asyncio.Condition是一个允许 async 函数在某些条件下等待或被通知，类似事件通知一样的。通常它被用于生产者-消费者模式之类的场景中，其中多个任务可能需要在某个条件成立时等待，当条件改变时被唤醒。如下一个简单的示例所示：
import asyncio
# 共享资源
shared_resource = 0
# 创建一个事件循环对象
loop = asyncio.get_event_loop()
# 创建一个条件变量对象
condition = asyncio.Condition()
# 协程A：增加共享资源的值
async def coro_a():
global shared_resource
async with condition:
# 等待条件满足
print("我是A，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！")
await condition.wait()
shared_resource += 1
print("我是A，我被通知了，可以操作shared_resource的值了，shared_resource的值增加了！",shared_resource)
print("我再执行一次通知看看？？")
# condition.notify()
async def coro_c():
global shared_resource
async with condition:
# 等待条件满足
print("我是C，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！")
await condition.wait()
shared_resource += 1
print("我是C，我被通知了，可以操作shared_resource的值了，shared_resource的值增加了！",shared_resource)
# 协程B：减少共享资源的值
async def coro_b():
global shared_resource
async with condition:
# 模拟一些耗时操作
print("我需要准备通知协程A可以操作shared_resource的值，不过我先自己操作一次先！！")
await asyncio.sleep(1)
shared_resource -= 1
print("协程B操作了一次，shared_resource的值减少了一次！",shared_resource)
# 通知等待条件的其他协程
condition.notify_all()
print("我已经通知协程A和C可以操作shared_resource的值了，通知完毕！")
async def main():
# 创建两个协程，一个等待事件，一个设置事件
await asyncio.gather(coro_a(),coro_b(),coro_c())
# 运行事件循环
asyncio.run(main())
# 启动协程A和协程B
# loop.create_task(coro_a())
# loop.create_task(coro_b())
# 运行事件循环
# loop.run_forever()
在上面的代码中，首先我们创建一个 asyncio.Condition 对象 condition，它将用于协程间的同步。接着定义一个全局变量 shared_resource 作为共享资源，初始值为 0。
然后定义协程 coro_a，它用于增加 shared_resource 的值。在这个协程中，首先使用 async with condition 语句获取条件变量的锁，（async with condition 语句是异步上下文管理器的使用方式，它用于获取条件变量的锁。在 asyncio 中，asyncio.Condition 实例同时也是一个异步上下文管理器，它可以与 async with 语句一起使用来自动管理锁的获取和释放，当 async with 块完全执行完毕后，锁会被自动释放，即使块内发生了异常也一样的会进行所的释放）。
然后打印一条消息表明它正在等待通知，接着调用 await condition.wait() 开始进入被的锁状态并等待通知，当 我们的coro_a 被通知时，执行解锁，将 shared_resource 的值增加 1，并打印输出一条消息表明操作已完成。
定义协程 coro_b，它将减少 shared_resource 的值。在这个协程中，同样使用 async with condition 获取锁，然后打印一条消息表明它正在准备通知 coro_a。接着，它使用 await asyncio.sleep(1) 模拟一些耗时操作，然后将 shared_resource 的值减少 1，并打印一条消息表明操作已完成。
在 oro_b中比较关键的地方就是，我们通过调用 condition.notify() 发出通知，告诉 coro_a 可以继续执行！
所以上面代码的输出结果为:
我是A，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！
我需要准备通知协程A可以操作shared_resource的值，不过我先自己操作一次先！！
协程B操作了一次，shared_resource的值减少了一次！ -1
我已经通知协程A和C可以操作shared_resource的值了，通知完毕！
我是C，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！
我是A，我被通知了，可以操作shared_resource的值了，shared_resource的值增加了！ 0
我再执行一次通知看看？？
但是上面的结果，你会发现，我们的事件循环会一直无法退出。问题的原因在于，我们的启动协程的时候的顺序性，首先我们上面的顺序是：
await asyncio.gather(coro_a(),coro_b(),coro_c())
其中我们的coro_b()是在第二个，也就是说会存在这样的一种情况： 当我们的coro_b()发出通知的时候，我们coro_a()已经开始被唤醒了，但是此时我们的，但是此时我们的coro_c()还没开始执行。当开始执行的时候就是重新进入的等待的被唤醒的状态。所以会出现一直等待的。 如果我们的调换一下启动顺序，如下：
await asyncio.gather(coro_a(),coro_c(),coro_b())
则我们输出的结果符合我们预期的效果：
我是A，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！
我是C，我需要等待协程B来通知我才可以操作shared_resource的值，等通知吧！
我需要准备通知协程A可以操作shared_resource的值，不过我先自己操作一次先！！
协程B操作了一次，shared_resource的值减少了一次！ -1
我已经通知协程A和C可以操作shared_resource的值了，通知完毕！
我是A，我被通知了，可以操作shared_resource的值了，shared_resource的值增加了！ 0
我再执行一次通知看看？？
我是C，我被通知了，可以操作shared_resource的值了，shared_resource的值增加了！ 1
相当于我们我们应该首先启动消费者（也就是coro_a(),coro_c()），然后再启动的我们生产者（发生事件通知）。
condition.notify(N)允许通知个数
condition.notify(1)是一个允许我们指定唤醒任务的数的一个方法，但是这个需要注意的地方就是，当我们调用 condition.notify(n) 时，将唤醒最多 n 个在条件变量上等待的任务。然而至于哪个任务会被先唤醒，这是由事件循环（event loop）决定的，并且不是固定的，它何时被调度取决于任务的调度策略。所以这里会存在一个顺序性的问题。因为事件循环的实现和调度策略可能会有所不同。在某些情况下，可能会使用优先级或其他策略来决定唤醒顺序。
如下的代码中，如果我们修改为如下的：
import asyncio
async def consumer(condition, n):
async with condition:
print(f"工作者 {n} 正在等待被唤醒...")
await condition.wait()
# 如果没有下面这个代码，程序会直接执行到这里，而不会等待condition.wait()，那么这个任务将一直等待
# condition.notify(1)
print(f"工作者 {n} 已经被唤醒")
async def producer(condition):
async with condition:
print("生产者 开始工作...")
await asyncio.sleep(1)
print("生产者 开始工作...通知一个任务唤醒-condition.notify(1)")
condition.notify(1)
# condition.notify_all()
print("生产者 已经工作...通某个任务唤醒-notify.notify_all()")
async def main():
condition = asyncio.Condition()
consumer_tasks = [asyncio.create_task(consumer(condition, i)) for i in range(2)]
producer_task = asyncio.create_task(producer(condition))
await asyncio.gather(*consumer_tasks, producer_task)
asyncio.run(main())
我们通过生产则发布 condition.notify(1)来调整其他协程的处理，如果我们只发出一个的话1，那整个会和前面的一样，所以我们需要进行当第一个处理完成，再发一个，这样才可以完整执行完成所有的任务。
asyncio.Semaphore
asyncio.Semaphore 是用于控制对共享资源的访问。它基于信号量（Semaphore）的概念，允许一定数量的并发访问。asyncio.Semaphore 通常用于限制并发执行的协程数量，以避免资源争抢和限制对共享资源的访问。如下示例所示：
import asyncio
async def worker(semaphore, worker_id):
"""
异步任务函数，模拟一个工人的工作过程。
参数：
semaphore: asyncio.Semaphore对象，用于控制并发访问的信号量。
worker_id: int，工人的编号。
返回值：
无返回值。
"""
print(f"工人{worker_id}正在等待获取信号量。")
async with semaphore:
print(f"工人{worker_id}已获取信号量，开始工作。")
await asyncio.sleep(1)
print(f"工人{worker_id}释放信号量。")
print(f"工人{worker_id}已完成工作。")
async def main():
"""
主函数，创建多个异步任务并等待它们完成。
参数：
无参数。
返回值：
无返回值。
"""
semaphore = asyncio.Semaphore(3) # 允许2个并发访问
tasks = [asyncio.create_task(worker(semaphore, i)) for i in range(5)]
await asyncio.gather(*tasks)
asyncio.run(main())
在上面示例中每个 worker 在开始工作前必须获取信号量，工作完成后释放信号量。如果所有可用的信号量都被获取，其他 worker 将被阻塞，直到有信号量被释放。所以当我们进行并发执行的时候，由于我们限制并发数，所以并发的5个协程不是同时，而是等待着semaphore的放行，才可以进行下一次批量的执行。
如下一个请求限量并发网络请求处理示例：
import asyncio
import aiohttp
import random
import logging
# 设置日志记录
logging.basicConfig(level=logging.INFO)
# 模拟资源池
class RequestResourcePool:
def __init__(self, size):
self.size = size
self.semaphore = asyncio.Semaphore(size)
self.resources = [None] * size
self.session = aiohttp.ClientSession() # 创建一个会话实例
async def acquire(self):
print(f"等待资源...")
await self.semaphore.acquire()
print(f"资源已获取!")
return self.semaphore
def release(self):
print(f"释放资源!")
self.semaphore.release()
async def fetch(self, url, task_id):
logging.info(f"任务 {task_id} 开始请求: {url}")
try:
async with self.session.get(url) as response:
response.raise_for_status() # 确保响应状态码为200
result = await response.text()
logging.info(f"任务 {task_id} 响应: {result}")
return result
except aiohttp.ClientError as e:
logging.error(f"任务 {task_id} 请求失败: {e}")
return None
async def action_request(self, url, task_id):
try:
sem = await self.acquire()
print(f"任务 {task_id} 开始使用资源")
# 发送HTTP请求
async with self.session.get(url) as response:
response.raise_for_status() # 确保响应状态码为200
result = await response.text()
print(f"任务 {task_id} 响应: {result}")
return result
except asyncio.CancelledError:
print(f"任务 {task_id} 被取消，释放资源")
self.release()
raise
finally:
self.release()
async def close(self):
if self.session:
await self.session.close() # 关闭会话
# 主函数
async def main():
# 创建一个资源池，限制为3个并发访问
pool = RequestResourcePool(3)
# 要请求的API列表
urls = [f"https://www.hao123.com/?{i}" for i in range(1, 11)]
# 创建并运行请求任务
tasks = [asyncio.create_task(pool.action_request(url, i)) for i, url in enumerate(urls, start=1)]
# 等待所有任务完成
await asyncio.gather(*tasks)
# 关闭资源池以释放资源
await pool.close()
# 运行事件循环
asyncio.run(main())
asyncio.Lock
asyncio.Lock 相对比较简单，和我们之前理解的锁是一样的，它用于控制对共享资源的访问。它其实和我们的 threading.Lock有效，但是它是为异步编程设计的。asyncio.Lock 允许 async 函数在某些条件下等待或被通知。如下一个示例：
import asyncio
# 异步协程，模拟一个工作协程
async def worker(lock, shared_resource):
async with lock:
# 打印当前工作协程正在访问共享资源
print(f"工作协程 {shared_resource} 正在访问共享资源。")
# 模拟一些耗时操作
await asyncio.sleep(1)
# 打印当前工作协程已完成
print(f"工作协程 {shared_resource} 已完成。")
# 异步主函数
async def main():
lock = asyncio.Lock()
# 创建并运行工作协程，共享资源分别为1、2、3
tasks = [asyncio.create_task(worker(lock, shared_resource)) for shared_resource in range(1, 4)]
# 等待所有任务完成
await asyncio.gather(*tasks)
# 启动异步主函数
asyncio.run(main())
在上面的示例中，我们本来如果没有锁的情况下，会并发进行协程任务的执行，但是加上锁了之后，整个任务的执行就是串行了，最终输出的结果如下：
工作协程 1 正在访问共享资源。
工作协程 1 已完成。
工作协程 2 正在访问共享资源。
工作协程 2 已完成。
工作协程 3 正在访问共享资源。
工作协程 3 已完成。
文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流哦！
结尾
END
简书：https://www.jianshu.com/u/d6960089b087
掘金：https://juejin.cn/user/2963939079225608
公众号：微信搜【程序员小钟同学】
新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
原文作者：程序员小钟同学；如有侵权，请联系管理员删除。
最新文章
- 深入FastAPI异步编程-之并发编程（15）-定义全局的支持全局上下文变量和局部上下文的对象可以任意传递了！
- 深入FastAPI异步编程-之并发编程（13）-纠错-修改基于aiojobs库结合aiojobs的和FastAPI框架的示例
- 深入FastAPI异步编程-之并发编程（13）-基于aiojobs库实现类似Background Tasks的效果
- 深入FastAPI异步编程-之并发编程（12）-基于pyee异步事件驱动通知
- 深入FastAPI异步编程-之并发编程（10）-?asyncio.wait 更精细化并发补充说明
- 深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明
- 深入FastAPI异步编程-之并发编程（8）-asyncio.gather的补充说明
- 深入FastAPI异步编程-之并发编程（7）番外篇~Windows系统下事件循环插曲
- 深入FastAPI异步编程-之并发编程（6）~事件循环
- 深入FastAPI异步编程-之并发编程（5）~Future对象
