同步原语¶
源代码: Lib/asyncio/locks.py
asyncio 同步原语的设计类似于 threading
模块,但有两个重要的注意事项
asyncio 原语不是线程安全的,因此不应将其用于 OS 线程同步(为此,请使用
threading
);这些同步原语的方法不接受 timeout 参数;使用
asyncio.wait_for()
函数执行带超时时间的操作。
asyncio 具有以下基本同步原语
锁¶
- class asyncio.Lock¶
为 asyncio 任务实现互斥锁。不是线程安全的。
asyncio 锁可用于保证对共享资源的独占访问。
使用锁的首选方法是
async with
语句lock = asyncio.Lock() # ... later async with lock: # access shared state
它等效于
lock = asyncio.Lock() # ... later await lock.acquire() try: # access shared state finally: lock.release()
3.10 版中已更改: 已删除 loop 参数。
- 协程 acquire()¶
获取锁。
此方法将等待锁 解锁,将其设置为 锁定,并返回
True
。当多个协程在
acquire()
中阻塞,等待锁解锁时,最终只有一个协程会继续执行。获取锁是 公平的:继续执行的协程将是第一个开始等待锁的协程。
- 释放()¶
释放锁。
当锁处于锁定状态时,将其重置为解锁状态并返回。
如果锁处于解锁状态,则会引发
RuntimeError
。
- 锁定()¶
如果锁处于锁定状态,则返回
True
。
事件¶
- 类 asyncio.事件¶
事件对象。非线程安全。
asyncio 事件可用于通知多个 asyncio 任务某些事件已发生。
事件对象管理一个内部标志,该标志可使用
set()
方法设置为true,并使用clear()
方法重置为false。wait()
方法会一直阻塞,直到标志设置为true。该标志最初设置为false。3.10 版中已更改: 已删除 loop 参数。
示例
async def waiter(event): print('waiting for it ...') await event.wait() print('... got it!') async def main(): # Create an Event object. event = asyncio.Event() # Spawn a Task to wait until 'event' is set. waiter_task = asyncio.create_task(waiter(event)) # Sleep for 1 second and set the event. await asyncio.sleep(1) event.set() # Wait until the waiter task is finished. await waiter_task asyncio.run(main())
- 设置()¶
设置事件。
所有等待事件设置的任务都将立即唤醒。
- 已设置()¶
如果事件已设置,则返回
True
。
条件¶
- 类 asyncio.Condition(lock=None)¶
条件对象。非线程安全。
任务可以使用 asyncio 条件原语等待某些事件发生,然后独占访问共享资源。
从本质上讲,条件对象结合了
Event
和Lock
的功能。多个条件对象可以共享一个锁,这允许对特定状态感兴趣的不同任务协调对共享资源的独占访问。可选的 lock 参数必须是
Lock
对象或None
。在后一种情况下,将自动创建一个新的锁对象。3.10 版中已更改: 已删除 loop 参数。
使用条件的首选方式是
async with
语句cond = asyncio.Condition() # ... later async with cond: await cond.wait()
它等效于
cond = asyncio.Condition() # ... later await cond.acquire() try: await cond.wait() finally: cond.release()
- 协程 acquire()¶
获取底层锁。
此方法将一直等到底层锁解锁,将其设置为锁定并返回
True
。
- notify(n=1)¶
唤醒最多 n 个(默认情况下为 1 个)在此条件上等待的任务。如果没有任务正在等待,则此方法为无操作。
在调用此方法之前必须获取锁,并在调用后立即释放锁。如果使用未锁定的锁调用,则会引发
RuntimeError
错误。
- locked()¶
如果获取了底层锁,则返回
True
。
- notify_all()¶
唤醒在此条件上等待的所有任务。
此方法的作用类似于
notify()
,但会唤醒所有正在等待的任务。在调用此方法之前必须获取锁,并在调用后立即释放锁。如果使用未锁定的锁调用,则会引发
RuntimeError
错误。
- release()¶
释放底层锁。
当在未锁定的锁上调用时,会引发
RuntimeError
。
- 协程 wait()¶
等待直到收到通知。
如果调用任务在调用此方法时未获取锁,则会引发
RuntimeError
。此方法释放基础锁,然后阻塞,直到通过
notify()
或notify_all()
调用唤醒它。唤醒后,Condition 重新获取其锁,此方法返回True
。
- 协程 wait_for(谓词)¶
等待直到谓词变为真。
谓词必须是一个可调用对象,其结果将被解释为一个布尔值。最终值是返回值。
信号量¶
- 类 asyncio.Semaphore(值=1)¶
一个信号量对象。非线程安全。
信号量管理一个内部计数器,该计数器由每次
acquire()
调用递减,由每次release()
调用递增。计数器绝不会低于零;当acquire()
发现它为零时,它会阻塞,等待某个任务调用release()
。可选的value参数为内部计数器提供初始值(默认值为
1
)。如果给定的值小于0
,则会引发ValueError
。3.10 版中已更改: 已删除 loop 参数。
使用信号量的首选方法是
async with
语句sem = asyncio.Semaphore(10) # ... later async with sem: # work with shared resource
它等效于
sem = asyncio.Semaphore(10) # ... later await sem.acquire() try: # work with shared resource finally: sem.release()
- locked()¶
如果无法立即获取信号量,则返回
True
。
- release()¶
释放信号量,将内部计数器加一。可以唤醒等待获取信号量的任务。
与
BoundedSemaphore
不同,Semaphore
允许进行比acquire()
调用更多的release()
调用。
BoundedSemaphore¶
- 类 asyncio.BoundedSemaphore(value=1)¶
有界信号量对象。非线程安全。
有界信号量是
Semaphore
的一个版本,如果在release()
中将内部计数器增加到超过初始value,则会引发ValueError
。3.10 版中已更改: 已删除 loop 参数。
Barrier¶
- 类 asyncio.Barrier(parties)¶
一个屏障对象。非线程安全的。
屏障是一个简单的同步原语,它允许阻塞,直到各方数量的任务在它上面等待。任务可以在
wait()
方法上等待,并且将被阻塞,直到指定数量的任务最终在wait()
上等待。在这一点上,所有等待的任务将同时解除阻塞。async with
可以用作等待wait()
的替代方案。屏障可以被重复使用任意次。
示例
async def example_barrier(): # barrier with 3 parties b = asyncio.Barrier(3) # create 2 new waiting tasks asyncio.create_task(b.wait()) asyncio.create_task(b.wait()) await asyncio.sleep(0) print(b) # The third .wait() call passes the barrier await b.wait() print(b) print("barrier passed") await asyncio.sleep(0) print(b) asyncio.run(example_barrier())
此示例的结果是
<asyncio.locks.Barrier object at 0x... [filling, waiters:2/3]> <asyncio.locks.Barrier object at 0x... [draining, waiters:0/3]> barrier passed <asyncio.locks.Barrier object at 0x... [filling, waiters:0/3]>
在 3.11 版本中添加。
- 协程 wait()¶
通过屏障。当屏障中的所有任务方都调用此函数时,它们将同时解除阻塞。
当屏障中的等待或阻塞任务被取消时,此任务将退出保持在相同状态的屏障。如果屏障的状态为“填充”,则等待任务的数量将减少 1。
返回值是 0 到
parties-1
范围内的整数,对于每个任务都不同。这可用于选择一个任务来执行一些特殊的维护,例如... async with barrier as position: if position == 0: # Only one task prints this print('End of *draining phase*')
如果在任务等待时屏障被破坏或重置,此方法可能会引发
BrokenBarrierError
异常。如果任务被取消,它可能会引发CancelledError
。
- 协程 reset()¶
将屏障返回到默认的空状态。在它上面等待的任何任务都将收到
BrokenBarrierError
异常。如果屏障被破坏,最好将其保留并创建一个新的屏障。
- 协程 abort()¶
将屏障置于损坏状态。这会导致对
wait()
的任何活动或未来调用都将失败,并出现BrokenBarrierError
。例如,如果其中一个任务需要中止,请使用此方法来避免无限等待任务。
- parties¶
通过屏障所需的任务数。
- n_waiting¶
填充时当前在屏障中等待的任务数。
- broken¶
如果屏障处于损坏状态,则为
True
的布尔值。
- exception asyncio.BrokenBarrierError¶
此异常是
RuntimeError
的子类,在Barrier
对象重置或损坏时引发。
3.9 版中已更改:使用 await lock
或 yield from lock
和/或 with
语句 (with await lock
,with (yield from lock)
) 获取锁已移除。请改用 async with lock
。