Asyncio 同步原语：非阻塞环境下的实现机制
大家好！今天我们来深入探讨 asyncio 库中的同步原语，包括 Lock、Semaphore 和 Event，重点分析它们在非阻塞环境下的实现机制。在传统的并发编程中，我们依赖线程和锁来保证共享资源的安全访问。但在 asyncio 的单线程事件循环模型下，直接使用线程锁会导致阻塞，破坏异步并发的特性。因此，asyncio 提供了一套专门设计的同步原语，它们基于协程和 future 实现，能够在不阻塞事件循环的情况下进行同步操作。
1. 为什么需要同步原语？
在 asyncio 中，多个协程并发执行，共享相同的事件循环和内存空间。如果没有适当的同步机制，多个协程同时访问和修改共享资源可能会导致数据竞争和不一致性。例如，考虑一个简单的计数器：
import asyncio
counter = 0
async def increment():
global counter
for _ in range(10000):
counter += 1
async def main():
await asyncio.gather(increment(), increment())
print(f"Counter value: {counter}")
if __name__ == "__main__":
asyncio.run(main())这段代码看似简单，但如果运行多次，你会发现 counter 的最终值并不总是 20000。这是因为两个 increment 协程可能同时读取 counter 的值，导致更新丢失。
因此，我们需要一种机制来协调多个协程对共享资源的访问，确保数据的一致性和正确性。asyncio 的同步原语就是为了解决这个问题而设计的。
2. Lock：互斥锁
Lock 是最基本的同步原语，用于实现互斥访问。它保证在任何时刻只有一个协程可以持有锁，从而防止多个协程同时访问临界区。
2.1 Lock 的基本用法
import asyncio
async def main():
lock = asyncio.Lock()
async def critical_section(name):
print(f"{name} trying to acquire lock")
async with lock: # 相当于 try: await lock.acquire(); finally: lock.release()
print(f"{name} acquired lock")
await asyncio.sleep(1) # 模拟耗时操作
print(f"{name} releasing lock")
await asyncio.gather(critical_section("Coroutine 1"), critical_section("Coroutine 2"))
if __name__ == "__main__":
asyncio.run(main())在这个例子中，我们创建了一个 asyncio.Lock 对象。critical_section 协程使用 async with lock: 语句来获取锁。async with 语句确保在协程执行完毕后自动释放锁，即使发生异常。
2.2 Lock 的实现机制
asyncio 的 Lock 基于 Future 实现。当一个协程尝试获取锁时，如果锁当前未被持有，则锁会被立即获取，并设置一个内部的 Future 对象为 "Done" 状态。如果锁已被持有，则协程会创建一个新的 Future 对象，并将该 Future 对象添加到锁的等待队列中。
当持有锁的协程释放锁时，它会检查等待队列是否为空。如果队列不为空，则它会从队列中取出一个 Future 对象，将其设置为 "Done" 状态，并允许对应的协程继续执行。如果队列为空，则锁会被标记为未持有。
下面是一个简化的 Lock 实现：
import asyncio
class Lock:
def __init__(self):
self._locked = False
self._waiters = []
async def acquire(self):
if not self._locked:
self._locked = True
return True
waiter = asyncio.Future()
self._waiters.append(waiter)
try:
await waiter
return True
except asyncio.CancelledError:
# Remove from waiters list on cancellation.
try:
self._waiters.remove(waiter)
except ValueError:
# If already removed, nothing to do.
pass
raise
def release(self):
if not self._locked:
raise RuntimeError("Lock is not acquired.")
if self._waiters:
waiter = self._waiters.pop(0)
if not waiter.done(): # Check if the waiter is cancelled
waiter.set_result(None) # Resolve the future
else:
self._locked = False
async def __aenter__(self):
await self.acquire()
async def __aexit__(self, exc_type, exc_val, exc_tb):
self.release()关键点：
_locked: 布尔值，表示锁是否被持有。_waiters: 一个队列，存储等待获取锁的 Future 对象。acquire(): 尝试获取锁。如果锁未被持有，则立即获取。否则，创建一个 Future 对象并添加到等待队列中，然后等待该 Future 对象变为 "Done" 状态。release(): 释放锁。如果等待队列不为空，则从队列中取出一个 Future 对象并将其设置为 "Done" 状态，否则将锁标记为未持有。__aenter__和__aexit__: 实现async with语句的上下文管理器协议。
2.3 Lock 如何实现非阻塞？
Lock 的非阻塞特性体现在 acquire() 方法中。当锁已被持有时，协程不会阻塞等待，而是创建一个 Future 对象并添加到等待队列中，然后 await waiter。await 关键字会将控制权交还给事件循环，允许其他协程继续执行。当锁被释放时，事件循环会唤醒等待队列中的某个协程，使其继续执行。
3. Semaphore：信号量
Semaphore 是一种更通用的同步原语，用于控制对有限数量资源的并发访问。与 Lock 只能允许一个协程访问资源不同，Semaphore 允许最多 N 个协程同时访问资源。
3.1 Semaphore 的基本用法
import asyncio
async def main():
semaphore = asyncio.Semaphore(3) # 允许最多 3 个协程同时访问
async def worker(name):
async with semaphore:
print(f"{name} acquired semaphore")
await asyncio.sleep(1)
print(f"{name} releasing semaphore")
await asyncio.gather(
worker("Worker 1"),
worker("Worker 2"),
worker("Worker 3"),
worker("Worker 4"),
worker("Worker 5"),
)
if __name__ == "__main__":
asyncio.run(main())在这个例子中，我们创建了一个 asyncio.Semaphore 对象，并将初始值设置为 3。这意味着最多可以有 3 个协程同时持有信号量。当一个协程尝试获取信号量时，如果信号量的值大于 0，则信号量的值减 1，协程可以继续执行。否则，协程会被添加到等待队列中，直到信号量的值大于 0。
3.2 Semaphore 的实现机制
Semaphore 的实现与 Lock 类似，也是基于 Future 实现。不同之处在于，Semaphore 维护一个内部的计数器，表示可用资源的数量。
下面是一个简化的 Semaphore 实现：
import asyncio
class Semaphore:
def __init__(self, value=1):
self._value = value
self._waiters = []
async def acquire(self):
while self._value <= 0:
waiter = asyncio.Future()
self._waiters.append(waiter)
try:
await waiter
except asyncio.CancelledError:
try:
self._waiters.remove(waiter)
except ValueError:
pass
raise
self._value -= 1
return True
def release(self):
self._value += 1
if self._waiters:
waiter = self._waiters.pop(0)
if not waiter.done():
waiter.set_result(None)
async def __aenter__(self):
await self.acquire()
async def __aexit__(self, exc_type, exc_val, exc_tb):
self.release()关键点：
_value: 整数，表示可用资源的数量。_waiters: 一个队列，存储等待获取信号量的 Future 对象。acquire(): 尝试获取信号量。如果_value大于 0，则将_value减 1 并返回。否则，创建一个 Future 对象并添加到等待队列中，然后等待该 Future 对象变为 "Done" 状态。release(): 释放信号量。将_value加 1。如果等待队列不为空，则从队列中取出一个 Future 对象并将其设置为 "Done" 状态。__aenter__和__aexit__: 实现async with语句的上下文管理器协议。
3.3 Semaphore 的应用场景
Semaphore 适用于需要限制并发访问数量的场景，例如：
- 限制对数据库连接池的并发访问。
- 限制对外部 API 的并发请求。
- 限制下载文件的并发数量。
4. Event：事件
Event 用于协程之间的通信。一个协程可以设置 Event 的状态，其他协程可以等待 Event 的状态变为已设置。
4.1 Event 的基本用法
import asyncio
async def main():
event = asyncio.Event()
async def waiter(name):
print(f"{name} waiting for event")
await event.wait()
print(f"{name} received event")
async def signaler():
await asyncio.sleep(2)
print("Setting event")
event.set()
await asyncio.gather(waiter("Waiter 1"), waiter("Waiter 2"), signaler())
if __name__ == "__main__":
asyncio.run(main())在这个例子中，我们创建了一个 asyncio.Event 对象。waiter 协程调用 event.wait() 方法等待事件被设置。signaler 协程在 2 秒后调用 event.set() 方法设置事件。
4.2 Event 的实现机制
Event 内部维护一个布尔标志，表示事件是否被设置。当事件未被设置时，调用 event.wait() 方法的协程会被添加到等待队列中。当事件被设置时，等待队列中的所有协程都会被唤醒。
下面是一个简化的 Event 实现：
import asyncio
class Event:
def __init__(self):
self._flag = False
self._waiters = []
def is_set(self):
return self._flag
async def wait(self):
if self._flag:
return
waiter = asyncio.Future()
self._waiters.append(waiter)
try:
await waiter
except asyncio.CancelledError:
try:
self._waiters.remove(waiter)
except ValueError:
pass
raise
def set(self):
self._flag = True
for waiter in self._waiters:
if not waiter.done():
waiter.set_result(None)
self._waiters = []
def clear(self):
self._flag = False关键点：
_flag: 布尔值，表示事件是否被设置。_waiters: 一个队列，存储等待事件被设置的 Future 对象。is_set(): 返回事件是否被设置。wait(): 等待事件被设置。如果事件已经被设置，则立即返回。否则，创建一个 Future 对象并添加到等待队列中，然后等待该 Future 对象变为 "Done" 状态。set(): 设置事件。将_flag设置为 True，并唤醒等待队列中的所有协程。clear(): 清除事件。将_flag设置为 False。
4.3 Event 的应用场景
Event 适用于协程之间的同步和通知，例如：
- 通知多个协程某个任务已经完成。
- 等待某个条件成立后再继续执行。
- 实现生产者-消费者模式。
5. 总结与对比
为了方便理解，我们用表格对这三种同步原语进行对比：
| 特性 | Lock | Semaphore | Event |
|---|---|---|---|
| 作用 | 互斥访问共享资源 | 控制对有限数量资源的并发访问 | 协程之间的同步和通知 |
| 允许并发数量 | 1 | N | 多个协程等待同一个事件 |
| 主要方法 | acquire(), release(), __aenter__, __aexit__ |
acquire(), release(), __aenter__, __aexit__ |
wait(), set(), clear(), is_set() |
| 内部机制 | 基于 Future 实现，维护一个锁标志和等待队列 | 基于 Future 实现，维护一个计数器和等待队列 | 基于 Future 实现，维护一个布尔标志和等待队列 |
6. 避免死锁
在使用同步原语时，需要特别注意避免死锁。死锁是指两个或多个协程相互等待对方释放资源，导致所有协程都无法继续执行的情况。
以下是一些避免死锁的常用技巧：
- 避免循环等待: 确保协程不会以循环的方式等待资源。
- 按固定顺序获取资源: 如果需要获取多个锁，尽量按照固定的顺序获取，避免不同的协程以不同的顺序获取锁。
- 使用超时: 在
acquire()方法中使用超时参数，如果超过指定时间仍未获取到锁，则放弃等待，避免永久阻塞。 - 使用 ContextVar 进行任务局部存储: 将一些状态信息存储在 ContextVar 中，避免多个协程之间共享状态，减少对锁的需求。
7. 最佳实践
- 尽可能减少锁的使用: 锁会引入额外的开销，降低程序的性能。尽量通过无锁算法或数据结构来避免锁的使用。
- 使用
async with语句:async with语句可以确保锁在协程执行完毕后自动释放，避免忘记释放锁导致死锁。 - 选择合适的同步原语: 根据实际需求选择合适的同步原语。例如，如果只需要互斥访问共享资源，则使用 Lock。如果需要控制对有限数量资源的并发访问，则使用 Semaphore。如果需要协程之间的同步和通知，则使用 Event。
- 仔细测试你的代码: 编写充分的单元测试和集成测试，确保你的代码在各种并发场景下都能正常工作。
总结：asyncio 同步原语的重要性
asyncio 的同步原语是构建可靠异步并发应用的关键。理解它们的实现机制和使用方法，能够帮助我们编写出高效、安全的代码。合理运用 Lock, Semaphore 和 Event，能够在保证数据一致性的前提下，充分发挥 asyncio 的并发优势。
更多IT精英技术系列讲座，到智猿学院