同步原语常用函数,在 Python阿希诺库中有哪些

发布时间：2023-03-20 11:18:11 所属栏目：教程来源：

导读：这篇文章主要讲解了“Python Asyncio库之同步原语常用函数有哪些”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Python Asyncio库之同步原语常用

这篇文章主要讲解了“Python Asyncio库之同步原语常用函数有哪些”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Python Asyncio库之同步原语常用函数有哪些”吧！

前记
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]: # <-- 通过async with语句来控制锁的粒度
share_data[i] = i
await asyncio.sleep(0)
print(i, share_data[i] == i)
async def sub_add(i):
async with lock_dict[i]:
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):
lock_dict[i] = asyncio.Lock()
task_list.append(sub(i))
task_list.append(sub_add(i))
await asyncio.gather(*task_list)
if __name__ == "__main__":
asyncio.run(main())
从例子可以看到asyncio.Lock的使用方法跟多线程的Lock差不多，通过async with语法来获取和释放锁，它的原理也很简单，主要做了如下几件事：

1.确保某一协程获取锁后的执行期间，别的协程在获取锁时需要一直等待，直到执行完成并释放锁。

2.当有协程持有锁的时候，其他协程必须等待，直到持有锁的协程释放了锁。

3.确保所有协程能够按照获取的顺序获取到锁。

这意味着需要有一个数据结构来维护当前持有锁的协程的和下一个获取锁协程的关系，同时也需要一个队列来维护多个获取锁的协程的唤醒顺序。

asyncio.Lock跟其它asyncio功能的用法一样，使用asyncio.Future来同步协程之间锁的状态，使用deque维护协程间的唤醒顺序，源码如下:

class Lockl(_ContextManagerMixin, mixins._LoopBoundMixin):
def __init__(self):
self._waiters = None
self._locked = False
def locked(self):
return self._locked
async def acquire(self):
if (not self._locked and (self._waiters is None or all(w.cancelled() for w in self._waiters))):
# 目前没有其他协程持有锁，当前协程可以运行
self._locked = True
return True
if self._waiters is None:
self._waiters = collections.deque()
# 创建属于自己的容器，并推送到`_waiters`这个双端队列中
fut = self._get_loop().create_future()
self._waiters.append(fut)
try:
try:
await fut
finally:
# 如果执行完毕，需要把自己移除，防止被`wake_up_first`调用
self._waiters.remove(fut)
except exceptions.CancelledError:
# 如果是等待的过程中被取消了，需要唤醒下一个调用`acquire`
if not self._locked:
self._wake_up_first()
raise
# 持有锁
self._locked = True
return True
def release(self):
if self._locked:
# 释放锁
self._locked = False
self._wake_up_first()
else:
raise RuntimeError('Lock is not acquired.')
def _wake_up_first(self):
if not self._waiters:
return
# 获取还处于锁状态协程对应的容器
try:
# 获取下一个等待获取锁的waiter
fut = next(iter(self._waiters))
except stopiteration:
return
# 设置容器为True,这样对应协程就可以继续运行了。
if not fut.done():
fut.set_result(True)
通过源码可以知道，锁主要提供了获取和释放的功能，对于获取锁需要区分两种情况：

1:当有协程想要获取锁时会先判断锁是否被持有，如果当前锁没有被持有就直接返回，使协程能够正常运行。

2:如果协程获取锁时，锁发现自己已经被其他协程持有则创建一个属于当前协程的asyncio.Future，用来同步状态，并添加到deque中。

而对于释放锁就比较简单，只要获取deque中的第一个asyncio.Future，并通过fut.set_result(True)进行标记，使asyncio.Future从peding状态变为done状态，这样一来，持有该asyncio.Future的协程就能继续运行，从而持有锁。

不过需要注意源码中acquire方法中对CancelledError异常进行捕获，再唤醒下一个锁，这是为了解决acquire方法执行异常导致锁一直被卡住的场景，通常情况下这能解决大部分的问题，但是如果遇到错误的封装时，我们需要亲自处理异常，并执行锁的唤醒。比如在通过继承asyncio.Lock编写一个超时锁时，最简单的实现代码如下:

import asyncio
class TimeoutLock(asyncio.Lock):
def __init__(self, timeout, *, loop=None):
self.timeout = timeout
super().__init__(loop=loop)
async def acquire(self) -> 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语法。

（编辑：驾考网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!