Python Asyncio 库之同步原语常用函数详解

您所在的位置：网站首页 › self__value › Python Asyncio 库之同步原语常用函数详解

Python Asyncio 库之同步原语常用函数详解

2023-04-02 16:20| 来源: 网络整理| 查看: 265

我正在参加「掘金·启航计划」

前记

Asyncio的同步原语可以简化我们编写资源竞争的代码和规避资源竞争导致的Bug的出现。但是由于协程的特性，在大部分业务代码中并不需要去考虑资源竞争的出现，导致Asyncio同步原语被使用的频率比较低，但是如果想基于Asyncio编写框架则需要学习同步原语的使用。

0.基础

同步原语都是适用于某些条件下对某个资源的争夺，在代码中大部分的资源都是属于一个代码块，而Python对于代码块的管理的最佳实践是使用with语法，with语法实际上是调用了一个类中的__enter__和__exit__方法，比如下面的代码：

class Demo(object): def __enter__(self): return def __exit__(self, exc_type, exc_val, exc_tb): return with Demo(): pass 复制代码

代码中的Demo类实现了__enter__和__exit__方法后，就可以被with语法调用，其中__enter__方法是进入代码块执行的逻辑，__enxi__方法是用于退出代码块(包括异常退出)的逻辑。这两个方法符合同步原语中对资源的争夺和释放，但是__enter__和__exit__两个方法都是不支持await调用的，为了解决这个问题，Python引入了async with语法。

async with语法和with语法类似，我们只要编写一个拥有__aenter__和__aexit__方法的类，那么这个类就支持asyncio with语法了，如下:

import asyncio class Demo(object): async def __aenter__(self): return async def __aexit__(self, exc_type, exc_val, exc_tb): return async def main(): async with Demo(): pass asyncio.run(main()) 复制代码

其中，类中的__aenter__方法是进入代码块时执行的方法，__aexit__是退出代码块时执行的方法。

有了async with语法的加持，asyncio的同步原语使用起来会比较方便，所以asyncio中对资源争夺的同步原语都会继承于_ContextManagerMixin类：

class _ContextManagerMixin: async def __aenter__(self): await self.acquire() # We have no use for the "as ..." clause in the with # statement for locks. return None async def __aexit__(self, exc_type, exc, tb): self.release() 复制代码

并实现了acquire和release方法，供__aenter__和__aexit__方法调用，同时我们在使用同步原语的时候尽量用到async with语法防止忘记释放资源的占用。

1.Lock

由于协程的特性，在编写协程代码时基本上可以不考虑到锁的情况，但在一些情况下我们还是需要用到锁，并通过锁来维护并发时的数据安全性，如下例子：

import asyncio share_data = {} async def sub(i): # 赋上相同的key和value share_data[i] = i await asyncio.sleep(0) print(i, share_data[i] == i) async def sub_add(i): # 赋上的value值是原来的+1 share_data[i] = i + 1 await asyncio.sleep(0) print(i, share_data[i] == i + 1) async def main(): # 创建并发任务 task_list = [] for i in range(10): task_list.append(sub(i)) task_list.append(sub_add(i)) # 并发执行 await asyncio.gather(*task_list) if __name__ == "__main__": asyncio.run(main()) 复制代码

在这个例子中程序会并发的执行sub和sub_add函数，他们是由不同的asyncio.Task驱动的，这意味着会出现这样一个场景。当负责执行sub(1)函数的asyncio.Task在执行完share_data[i]=i后就执行await asyncio.sleep(0)从而主动让出控制权并交还给事件循环，等待事件循环的下一次调度。不过事件循环不会空下来，而是马上安排下一个asyncio.Task执行，此时会先执行到sub_add(1)函数的share_data[i] = i + 1，并同样的在执行到await asyncio.sleep(0)的时候把控制权交会给事件循环。这时候控制权会由事件循环转移给原先执行sub(1)函数的asyncio.Task，获取到控制权l后sub(1)函数的逻辑会继续走，但由于share_data[i]的数据已经被share_data[i] = i + 1修改了，导致最后执行print时，share_data[i]的数据已经变为脏数据，而不是原本想要的数据了。

为了解决这个问题，我们可以使用asyncio.Lock来解决资源的冲突，如下:

import asyncio share_data = {} # 存放对应资源的锁 lock_dict = {} async def sub(i): async with lock_dict[i]: # bool: return await asyncio.wait_for(super().acquire(), self.timeout) 复制代码

这份代码非常简单，他只需要在__init__方法传入timeout参数，并在acuiqre方法中通过wait_for来实现锁超时即可，现在假设wait_for方法是一个无法传递协程cancel的方法，且编写的acquire没有进行捕获异常再释放锁的操作，当异常发生的时候会导致锁一直被卡住。为了解决这个问题，只需要对TimeoutLock的acquire方法添加异常捕获，并在捕获到异常时释放锁即可，代码如下:

class TimeoutLock(asyncio.Lock): def __init__(self, timeout, *, loop=None): self.timeout = timeout super().__init__(loop=loop) async def acquire(self) -> bool: try: return await asyncio.wait_for(super().acquire(), self.timeout) except Exception: self._wake_up_first() raise 复制代码 2.Event

asyncio.Event也是一个简单的同步原语，但它跟asyncio.Lock不一样，asyncio.Lock是确保每个资源只能被一个协程操作，而asyncio.Event是确保某个资源何时可以被协程操作，可以认为asyncio.Lock锁的是资源，asyncio.Event锁的是协程，所以asyncio.Event并不需要acquire来锁资源，release释放资源，所以也用不到async with语法。

asyncio.Event的简单使用示例如下：

import asyncio async def sub(event: asyncio.Event) -> None: await event.wait() print("I'm Done") async def main() -> None: event = asyncio.Event() for _ in range(10): asyncio.create_task(sub(event)) await asyncio.sleep(1) event.set() asyncio.run(main()) 复制代码

在这个例子中会先创建10个asyncio.Task来执行sub函数，但是所有sub函数都会在event.wait处等待，直到main函数中调用event.set后，所有的sub函数的event.wait会放行，使sub函数能继续执行。

可以看到asyncio.Event功能比较简单，它的源码实现也很简单，源码如下：

class Event(mixins._LoopBoundMixin): def __init__(self): self._waiters = collections.deque() self._value = False def is_set(self): return self._value def set(self): if not self._value: # 确保每次只能set一次 self._value = True # 设置每个协程存放的容器为True，这样对应的协程就可以运行了 for fut in self._waiters: if not fut.done(): fut.set_result(True) def clear(self): # 清理上一次的set self._value = False async def wait(self): if self._value: # 如果设置了，就不需要等待了 return True # 否则需要创建一个容器，并需要等待容器完成 fut = self._get_loop().create_future() self._waiters.append(fut) try: await fut return True finally: self._waiters.remove(fut) 复制代码

通过源码可以看到wait方法主要是创建了一个asyncio.Future，并把它加入到deque队列后就一直等待着，而set方法被调用时会遍历整个deque队列，并把处于peding状态的asyncio.Future设置为done，这时其他在调用event.wait方法的协程就会得到放行。

通过源码也可以看出，asyncio.Event并没有继承于_ContextManagerMixin，这是因为它锁的是协程，而不是资源。

asyncio.Event的使用频率比asyncio.Lock多许多，不过通常都会让asyncio.Event和其他数据结构进行封装再使用，比如实现一个服务器的优雅关闭功能，这个功能会确保服务器在等待n秒后或者所有连接都关闭后才关闭服务器，这个功能就可以使用set与asyncio.Event结合，如下:

import asyncio class SetEvent(asyncio.Event): def __init__(self, *, loop=None): self._set = set() super().__init__(loop=loop) def add(self, value): self._set.add(value) self.clear() def remove(self, value): self._set.remove(value) if not self._set: self.set() 复制代码

这个SetEvent结合了set和SetEvent的功能，当set有数据的时候，会通过clear方法使SetEvent变为等待状态，而set没数据的时候，会通过set方法使SetEvent变为无需等待的状态，所有调用wait的协程都可以放行，通过这种结合，SetEvent拥有了等待资源为空的功能。接下来就可以用于服务器的优雅退出功能：

async def mock_conn_io() -> None: await asyncio.sleep(1) def conn_handle(set_event: SetEvent): task: asyncio.Task = asyncio.create_task(mock_conn_io()) set_event.add(task) task.add_done_callback(lambda t: set_event.remove(t)) async def main(): set_event: SetEvent = SetEvent() for _ in range(10): conn_handle(set_event) # 假设这里收到了退出信号 await asyncio.wait(set_event.wait(), timeout=9) asyncio.run(main()) 复制代码

在这个演示功能中，mock_conn_io用于模拟服务器的连接正在处理中，而conn_handle用于创建服务器连接，main则是先创建10个连接，并模拟在收到退出信号后等待资源为空或者超时才退出服务。

这只是简单的演示，实际上的优雅关闭功能要考虑的东西不仅仅是这些。

4.Condition

condition只做简单介绍

asyncio.Condition是同步原语中使用最少的一种，因为他使用情况很奇怪，而且大部分场景可以被其他写法代替，比如下面这个例子:

import asyncio async def task(condition, work_list): await asyncio.sleep(1) work_list.append(33) print('Task sending notification...') async with condition: condition.notify() async def main(): condition = asyncio.Condition() work_list = list() print('Main waiting for data...') async with condition: _ = asyncio.create_task(task(condition, work_list)) await condition.wait() print(f'Got data: {work_list}') asyncio.run(main()) # >>> Main waiting for data... # >>> Task sending notification... # >>> Got data: [33] 复制代码

在这个例子中可以看到，notify和wait方法只能在async with condition中可以使用，如果没有在async with condition中使用则会报错，同时这个示例代码有点复杂，没办法一看就知道执行逻辑是什么，其实这个逻辑可以转变成一个更简单的写法：

import asyncio async def task(work_list): await asyncio.sleep(1) work_list.append(33) print('Task sending notification...') return async def main(): work_list = list() print('Main waiting for data...') _task = asyncio.create_task(task(work_list)) await _task print(f'Got data: {work_list}') asyncio.run(main()) # >>> Main waiting for data... # >>> Task sending notification... # >>> Got data: [33] 复制代码

通过这个代码可以看到这个写法更简单一点，而且更有逻辑性，而condition的写法却更有点Go协程写法/或者回调函数写法的感觉。所以建议在认为自己的代码可能会用到asyncio.Conditon时需要先考虑到是否需要asyncio.Codition？是否有别的方案代替，如果没有才考虑去使用asyncio.Conditonk。

5.Semaphore

asyncio.Semaphore--信号量是同步原语中被使用最频繁的，大多数都是用在限流场景中，比如用在爬虫中和客户端网关中限制请求频率。

asyncio.Semaphore可以认为是一个延缓触发的asyncio.Lock，asyncio.Semaphore内部会维护一个计数器，无论何时进行获取或释放，它都会递增或者递减(但不会超过边界值)，当计数器归零时，就会进入到锁的逻辑，但是这个锁逻辑会在计数器大于0的时候释放j，它的用法如下：`

import asyncio async def main(): semaphore = asyncio.Semaphore(10): async with semaphore: pass asyncio.run(main()) 复制代码

示例中代码通过async with来指明一个代码块（代码用pass代替），这个代码块是被asyncio.Semaphore管理的，每次协程在进入代码块时，asyncio.Semaphore的内部计数器就会递减一，而离开代码块则asyncio.Semaphore的内部计数器会递增一。当有一个协程进入代码块时asyncio.Semaphore发现计数器已经为0了，则会使当前协程进入等待状态，直到某个协程离开这个代码块时，计数器会递增一，并唤醒等待的协程，使其能够进入代码块中继续执行。

asyncio.Semaphore的源码如下，需要注意的是由于asyncio.Semaphore是一个延缓的asyncio.Lock，所以当调用一次release后可能会导致被唤醒的协程和刚进入代码块的协程起冲突，所以在acquire方法中要通过一个while循环来解决这个问题：`

class Semaphore(_ContextManagerMixin, mixins._LoopBoundMixin): def __init__(self, value=1): if value < 0: raise ValueError("Semaphore initial value must be >= 0") self._value = value self._waiters = collections.deque() self._wakeup_scheduled = False def _wake_up_next(self): while self._waiters: # 按照放置顺序依次弹出容器 waiter = self._waiters.popleft() if not waiter.done(): # 设置容器状态，使对应的协程可以继续执行 waiter.set_result(None) # 设置标记 self._wakeup_scheduled = True return def locked(self): return self._value == 0 async def acquire(self): # 如果`self._wakeup_scheduled`为True或者value小于0 while self._wakeup_scheduled or self._value asyncio.Task: await self._semaphore.acquire() task: asyncio.Task = asyncio.create_task(coro) task.add_done_callback(lambda t: self._semaphore.release()) return task async def demo(cnt: int) -> None: print(f"{int(time.time())} create {cnt} task...") await asyncio.sleep(cnt) async def main() -> None: pool: Pool = Pool(3) for i in range(10): await pool.create_task(demo(i)) asyncio.run(main()) # >>> 1677517996 create 0 task... # >>> 1677517996 create 1 task... # >>> 1677517996 create 2 task... # >>> 1677517996 create 3 task... # >>> 1677517997 create 4 task... # >>> 1677517998 create 5 task... # >>> 1677517999 create 6 task... # >>> 1677518001 create 7 task... # >>> 1677518003 create 8 task... # >>> 1677518005 create 9 task... 复制代码

【本文地址】

Python Asyncio 库之同步原语常用函数详解

Python Asyncio 库之同步原语常用函数详解

今日新闻

推荐新闻