2022年 11月 3日

python携程详解_python协程

python协程

协程本质上就是一个线程，不过它是协作式的非抢占式的程序，面向的是IO操作。python有GIL的限制，不能充分利用多线程的实现高并发。进程和线程都是通过cpu的调度实现不同任务的有序执行，协程则要在写代码的时候确定执行顺序。由于协程在一个线程中，所以协程不能阻塞。

优缺点：

无需线程上下文切换的开销

在一个线程中，不需要加锁

无法利用多核资源：协程的本质是单线程，需要和进程配合才能运行在多CPU上

进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序

python协程的发展时间较长：

python2.5 为生成器引用.send()、.throw()、.close()方法

python3.3 为引入yield from，可以接收返回值，可以使用yield from定义协程

Python3.4 加入了asyncio模块

Python3.5 增加async、await关键字，在语法层面的提供支持

python3.7 使用async def + await的方式定义协程

此后asyncio模块更加完善和稳定，对底层的API进行的封装和扩展

python将于 3.10版本中移除 以yield from的方式定义协程 (目前版本是3.9.1)

由于asyncio每个版本都会新增功能，对一些旧的底层的API进行封装，极大地方便的使用者，但正因为此，网上有很多教程使用的接口官方已经不建议直接使用，应该改而使用更加高级的API，所以在这里记录一下如何使用这些API。

简单例子

要点：

使用async def的形式定义

在协程中可以使用await关键字，注意其后跟的是”可等待对象”(协程, 任务 和 Future)

协程不能直接执行，需要在asyncio.run()中执行，也可以跟在await后面

async和await这两个关键字只能在协程中使用

import asyncio

async def foo(name):

await asyncio.sleep(1) # 这是一个不会阻塞的sleep，是一个协程

print(f”name = {name}”)

async def main():

# 协程本身就是一个可等待对象

await foo(“lczmx”) # 执行协程

print(“done”)

if __name__ == ‘__main__’:

# 使用asyncio.run运行

asyncio.run(main())

asyncio.run(main, *, debug=False)方法就是对run_until_complete进行了封装：

loop = events.new_event_loop()

return loop.run_until_complete(main)

关于可等待对象

可等待对象(awaitable)是能在 await 表达式中使用的对象。可以是 协程 或是具有__await__() 方法的对象。

那么协程是如何成为可等待对象的呢？

collections.abc.Awaitable类，这是为可等待对象提供的类，可被用于 await 表达式中

class Awaitable(metaclass=ABCMeta):

__slots__ = ()

@abstractmethod

def __await__(self):# __await__方法必须返回一个 iterator

yield

@classmethod

def __subclasshook__(cls, C):

if cls is Awaitable:

return _check_methods(C, “__await__”)

return NotImplemented

用async def复合语句创建的函数，它返回的是一个Coroutine对象，而Coroutine继承Awaitable。

并发

使用协程进行并发操作

方法一

使用asyncio.create_task(coro)方法，返回一个Task对象，Task类继承Future，在python3.7以下版本中使用asyncio.ensure_future(coro_or_future)。

import asyncio

async def foo(char:str, count: int):

for i in range(count):

print(f”{char}-{i}”)

await asyncio.sleep(.5)

async def main():

task1 = asyncio.create_task(foo(“A”, 2))

task2 = asyncio.create_task(foo(“B”, 3))

task3 = asyncio.create_task(foo(“C”, 2))

await task1

await task2

await task3

if __name__ == ‘__main__’:

asyncio.run(main())

执行结果

A-0

B-0

C-0

A-1

B-1

C-1

B-2

方法二

使用asyncio.gather()方法，其内部调用的是asyncio.ensure_future()方法

import asyncio

async def foo(char:str, count: int):

for i in range(count):

print(f”{char}-{i}”)

await asyncio.sleep(.5)

async def main():

await asyncio.gather(foo(“A”, 2), foo(“B”, 3), foo(“C”, 2))

if __name__ == ‘__main__’:

asyncio.run(main())

回调、返回值等操作

要完成这些功能需要Task对象，即asyncio.create_task()的返回值。由于Task继承Future，实现了除Future.set_result() 和 Future.set_exception()外的全部API，而asyncio.Future模仿的是 concurrent.futures.Future类，所以Task很多方法和 在使用线/进程池时用到的方法类似(有细微差别)。

Task的方法，见官方文档

使用回调函数和取得返回值的例子：

import asyncio

def callback(future):

# 唯一参数是一个Task对象

# print(type(future)) #

print(future)

# result=123>

print(future.result()) # 123 # 接收返回值

print(future.get_name()) # foo

async def foo():

print(“running”)

return 123

async def main():

task = asyncio.create_task(foo(), name=”foo”) # name形参3.8及以上版本可用

task.add_done_callback(callback) # 添加回调函数

await task

if __name__ == ‘__main__’:

asyncio.run(main())

与线程结合

我们知道，协程本身就只有一个线程，假如这协程阻塞了，那么整个程序也就阻塞了。为此我们在执行一些必然会产生阻塞的代码时，可以把代码放入到其它线程/进程中，这样可以继续执行协程的其它代码了。

方法一

coroutine asyncio.to_thread(func, /, *args, **kwargs)

这是python3.9的新方法，3.9以下版本看方法二

在不同的线程中异步地运行函数 func。向此函数提供的任何*args 和 **kwargs 会被直接传给 func。其返回值是一个协程，所以假如有回调等操作，使用asyncio.create_task(coro)方法，再调用Task对象的方法。

import asyncio

import time

def block_func(name: str):

time.sleep(2) # 模拟阻塞时间

print(f”name = {name}”)

async def foo():

# 一个协程

print(“async foo”)

await asyncio.sleep(1)

async def main():

await asyncio.gather(

asyncio.to_thread(block_func, name=”lczmx”),

foo()

)

if __name__ == ‘__main__’:

asyncio.run(main())

方法二

awaitable loop.run_in_executor(executor, func, *args)

安排在指定的执行器(线/进程池)中调用 func。该方法的返回值awaitable对象，其实就是一个asyncio.Future对象。这个方法使用起来也比较简单，不过要注意传参方式：位置参数可以直接传入，而关键字参数需要使用functools.partial()。

from concurrent.futures import ThreadPoolExecutor

from functools import partial

import asyncio

import time

def block_func(name: str, age: int):

time.sleep(2) # 模拟阻塞时间

print(f”name = {name}, age = {age}”)

async def foo():

# 一个协程

print(“async foo”)

await asyncio.sleep(1)

async def main():

loop = asyncio.get_running_loop()

with ThreadPoolExecutor(5) as pool:

task = loop.run_in_executor(pool, partial(block_func, “lczmx”, age=18))

# task可以添加回调等操作

await asyncio.gather(foo(), task)

if __name__ == ‘__main__’:

asyncio.run(main())

这个两个方法的关系

asyncio.to_thread()方法实质上就是对loop.run_in_executor()方法进行了封装：

async def to_thread(func, /, *args, **kwargs):

loop = events.get_running_loop()

ctx = contextvars.copy_context()

func_call = functools.partial(ctx.run, func, *args, **kwargs)

return await loop.run_in_executor(None, func_call)

假如B线程想要把协程c放入到A线程的事件循环中执行，使用syncio.run_coroutine_threadsafe(coro_func(), loop)方法，具体怎么使用见范例

异步上下文管理器

异步上下文管理器使用的是async with语法， 是 上下文管理器 的一种，它能够在其__aenter__ 和 __aexit__ 方法中暂停执行。使用异步上下文管理器之前，可以先看一看一般的上下文管理器，类比过来。

__aenter__(self)

在语义上类似于 __enter__()，仅有的区别是它必须返回一个 可等待对象。

__aexit__(self, exc_type, exc_value, traceback)

在语义上类似于 __exit__()，仅有的区别是它必须返回一个 可等待对象。

一个简单的例子：

import asyncio

class AContext:

def __init__(self):

print(“init running”)

async def __aenter__(self):

print(“aenter running”)

async def __aexit__(self, exc_type, exc_val, exc_tb):

print(“aexit running”)

async def main():

async with AContext() as ac:

print(“AContext”, ac) # AContext None # __aenter__ 没有return东西

if __name__ == ‘__main__’:

print(“start”)

asyncio.run(main())

使用标准库

除了上述方法可以实现异步上下文管理器外，还可以使用contextlib.asynccontextmanager装饰器+yield实现，yield前面的代码对应__aenter__，其后的代码对应__aexit__。

import contextlib

import asyncio

# 加上装饰器

@contextlib.asynccontextmanager

async def foo():

try:

# 进行初始化

yield “返回你要操作的对象”

finally:

# 处理释放资源等操作

pass

async def main():

async with foo() as f:

print(f)

if __name__ == “__main__”:

asyncio.run(main())

异步迭代器

异步迭代器使用的时async for语法，和常规的for表达式一样， async for也有一个可选的else 分句。异步迭代器有以下几个要求：

必须实现__aiter__方法，该方法返回一个异步迭代器对象。

异步迭代器对象必须实现__anext__方法，该方法返回一个awaitable类型的值。

为了停止迭代，__anext__必须抛出一个StopAsyncIteration异常。

跟迭代器协议非常类似。

例子：

class AsyncIteratorWrapper:

def __init__(self, obj):

self._it = iter(obj)

def __aiter__(self):

return self

async def __anext__(self):

try:

value = next(self._it)

except StopIteration:

raise StopAsyncIteration

return value

async for letter in AsyncIteratorWrapper(“abc”):

print(letter)

综合例子

非阻塞爬虫

from contextlib import asynccontextmanager

from concurrent.futures import ThreadPoolExecutor

from functools import partial

import requests

import asyncio

import time

class ASyncRequests:

“””非阻塞的requests”””

def __init__(self, count):

“”” :param count: 线程池大小 “””

assert isinstance(count, int)

self.pool = ThreadPoolExecutor(count) # 创建线程池

self.loop = asyncio.get_running_loop() # 获得事件循环

if not self.loop:

raise RuntimeError(“event loop为None”)

# self.all_future 用于装run_in_executor返回的future对象

# python3.9版本用to_thread

self.all_future = []

def get(self, url, callback, **kwargs):

self.requests(url, “get”, callback, **kwargs)

def post(self, url, callback, **kwargs):

self.requests(url, “post”, callback, **kwargs)

def requests(self, url, method, callback, **kwargs):

“”” 写了两种方法，第一种适合版本python，另一种适合最新版的python “””

# python3.5及以上版本可用

# run_in_executor:

# 把阻塞的函数放入线程池中执行

# 返回一个asyncio.Future对象

# 传值时，利用 偏函数 传关键字参数

result = self.loop.run_in_executor(self.pool, partial(requests.request,

url=url, method=method, **kwargs))

result.add_done_callback(callback) # 回调

self.all_future.append(result)

# python3.9及以上版本可用

# to_thread:

# 使用其它的线程，执行阻塞函数

# 返回一个协程

# 除第一个参数外的所有参数都会给func

# coro = asyncio.to_thread(requests.request, url=url, method=method, **kwargs)

# task = asyncio.create_task(coro)

# task.add_done_callback(callback) # 回调

# self.all_future.append(task)

def process_get(future):

“””回调函数，处理返回值”””

print(type(future)) #

# future.result()的返回值就是Response对象

print(future.result().status_code) # 200

@asynccontextmanager

async def request_session(count=5):

try:

async_req = ASyncRequests(count)

yield async_req

finally:

await asyncio.gather(*async_req.all_future)

async def main():

async with request_session(10) as session:

session.get(“https://www.baidu.com/”, callback=process_get)

session.get(“https://www.sogo.com/”, callback=process_get)

session.get(“https://cn.bing.com/”, callback=process_get)

session.get(“https://www.cnblogs.com/lczmx”, callback=process_get)

def block_req():

requests.get(“https://www.baidu.com/”)

requests.get(“https://www.sogo.com/”)

requests.get(“https://cn.bing.com/”)

requests.get(“https://www.cnblogs.com/lczmx”)

if __name__ == ‘__main__’:

start = time.time()

asyncio.run(main())

end = time.time()

print(“使用协程加线程：总用时: %f sec” % (end – start))

block_req()

print(“直接访问：总用时: %f sec” % (time.time() – end))

# 访问数越多，协程的优势就越大

# 访问同一个网站过多可能会报错 Connection aborted.’, ConnectionResetError(10054, ‘远程主机强迫关闭了一个现有的连接。

推荐阅读：

发表评论