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。

简单例子

要点：

1. 使用

   async def

   的形式定义
2. 在协程中可以使用

   await

   关键字，注意其后跟的是"可等待对象"(协程, 任务 和 Future)
3. 协程不能直接执行，需要在

   asyncio.run()

   中执行，也可以跟在

   await

   后面

4. 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

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

__await__()

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

1. 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

2. 用

   async def

   复合语句创建的函数，它返回的是一个

   Coroutine对象

   ，而

   Coroutine

   继承

   Awaitable

   。

并发

使用协程进行并发操作
方法一
使用

asyncio.create_task(coro)

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()

Future.set_result()

Future.set_exception()

concurrent.futures.Future

Task的方法，见官方文档
使用回调函数和取得返回值的例子：

import asyncio

def callback(future):
# 唯一参数是一个Task对象
# print(type(future))     # <class '_asyncio.Task'>

print(future)
# <Task finished name='Task-2' coro=<foo() done, defined at E: ... xxx.py:11> 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)

*args

**kwargs

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)

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(
ad8
)

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)

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

1. 必须实现

   __aiter__

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

   __anext__

   方法，该方法返回一个

   awaitable

   类型的值。
3. 为了停止迭代，

   __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))  # <class '_asyncio.Future'>

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

@asynccontextmanager
async def request_session(count=5):
try:
async_req = ASyncRequests(count)
y
e08
ield 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))
# 访问数越
1ca2
多，协程的优势就越大
# 访问同一个网站过多可能会报错 Connection aborted.', ConnectionResetError(10054, '远程主机强迫关闭了一个现有的连接。

推荐阅读：