同步原语

源代码: 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()方法重置为falsewait()方法会一直阻塞,直到标志设置为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。否则,一直阻塞,直到另一个任务调用set()

设置()

设置事件。

所有等待事件设置的任务都将立即唤醒。

清除()

清除(取消设置)事件。

等待wait()的任务现在会一直阻塞,直到再次调用set()方法。

已设置()

如果事件已设置,则返回True

条件

asyncio.Condition(lock=None)

条件对象。非线程安全。

任务可以使用 asyncio 条件原语等待某些事件发生,然后独占访问共享资源。

从本质上讲,条件对象结合了 EventLock 的功能。多个条件对象可以共享一个锁,这允许对特定状态感兴趣的不同任务协调对共享资源的独占访问。

可选的 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()
协程 acquire()

获取信号量。

如果内部计数器大于零,则将其减一并立即返回True。如果为零,则等待调用release(),然后返回True

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 lockyield from lock 和/或 with 语句 (with await lockwith (yield from lock)) 获取锁已移除。请改用 async with lock