asyncio 并发编程（一）

痛定思痛。 2023-09-29 21:17 2阅读 0赞

`Python2` 时代高性能的网络编程主要是 `Twisted`、`Tornado` 和 `Gevent` 这三个库，但是它们的异步代码相互之间既不兼容也不能移植。`Gvanrossum` 希望在 `Python 3` 实现一个原生的基于**生成器的协程库**，其中直接内置了对异步 IO 的支持，这就是 `asyncio`，它在 `Python 3.4` 被引入到标准库。

### 并发对比 ###

`asyncio` 使用**单线程**、**单个进程**的方式切换（通常程序等待读或写数据时就是切换上下文的时机）。

#### requests + ThreadPoolExecutor ####

这里采用的是线程池 + 同步方式，`requests` 仅支持同步方式：

import time
    
    import requests
    from concurrent.futures import ThreadPoolExecutor
    
    numbers = range(12)
    url = 'http://httpbin.org/get?a={}'
    
    
    def fetch(a):
        print(a)
        r = requests.get(url.format(a))
    
        return r.json()['args']['a']
    
    
    start = time.time()
    
    # 开三个线程
    with ThreadPoolExecutor(max_workers=3) as executor:
        for num, result in zip(numbers, executor.map(fetch, numbers)):
            print('fetch({})'.format(num, result))
    
    print('使用 requests + ThreadPoolExector 总耗时：{}'.format(time.time() - start))     # 5.696202754974365

#### asyncio + aiohttp ####

协程+异步，现在的 `asyncio`，有了很多的模块已经在支持`：aiohttp,aiodns,aioredis` 等等 `https://github.com/aio-libs`：

import asyncio
    import time
    import aiohttp
    
    url = 'http://httpbin.org/get?a={}'
    numbers = range(12)
    
    
    async def fetch(a):
        async with aiohttp.ClientSession() as session:
            async with session.get(url.format(a)) as resp:
                data = await resp.json()	# 等待结果
                return data['args']['a']
    
    
    start = time.time()
    event_loop = asyncio.get_event_loop()	# 新建事件循环
    tasks = [fetch(num) for num in numbers]	# 添加到任务列表
    
    # asyncio.gather() 按顺序搜集异步任务执行的结果
    results = event_loop.run_until_complete(asyncio.gather(*tasks))		# 开启事件循环
    
    for num, result in zip(numbers, results):
        print('fetch({}) = {}'.format(num, result))
    
    print('使用 asyncio + aiohttp 总耗时：{}'.format(time.time() - start))     # 1.5458192825317383

在想进行协程切换的地方使用 `await` 关键字，上述 `await r.json()` 会等待 `I/O`切换。可以看到**协程+异步的方式**比**同步+多线程**方法整整快了三四倍。

### asyncio 的基本使用 ###

协程本质上是异步非阻塞技术，它可以用一组少量的线程来实现多个任务，一旦某个任务阻塞，则可能用同一线程继续运行其他任务，避免大量上下文的切换，Python 中使用 `asyncio` 标准库实现协程。

关于 `asyncio` 中的一些关键字：

*  `event_loop` 事件循环：程序开启一个无限循环，将一些函数注册到事件循环中，当满足事件发生时，调用相应的协程函数
 *  `coroutine` 协程：协程对象，指一个使用 `async` 关键字定义的函数，它的调用不会立即执行函数，而是会返回一个协程对象。协程对象需要注册到事件循环，由事件循环调用。
 *  `task` 任务：一个协程对象就是一个原生可以挂起的函数，任务则是对协程进一步封装，其中包含了任务的各种状态
 *  `future`: 代表将来执行或没有执行的任务的结果。它和task上没有本质上的区别
 *  `async/await` 关键字：`python3.5` `用于定义协程的关键字，async` 定义一个协程，`await` 用于挂起阻塞的异步调用接口。
 *  事件循环：一种处理多并发量的机制，我们可以定义事件循环来简化使用轮询方法来监控事件

#### 快速入门 ####

import asyncio
    
    
    async def foo():
        print('这是一个协程')
    
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()         # 定义 loop 对象，事件循环
        try:
            print('开始运行协程')
            coro = foo()
            print('进入事件循环')
            loop.run_until_complete(coro)       # 用协程启动事件循环，协程返回，这个方法停止循环，接受一个 future 对象
        finally:
            print('关闭事件循环')
            loop.close()

运行结果：

开始运行协程
    进入事件循环
    这是一个协程
    关闭事件循环

#### 协程返回值 ####

import asyncio
    
    
    async def foo():
        print('这是一个协程')
        return '返回值'
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        try:
            print('开始运行协程')
            coro = foo()
            print('进入事件循环')
            result = loop.run_until_complete(coro)
            print('协程返回值：', result)
        finally:
            print('关闭事件循环')
            loop.close()

> **Tips：`run_until_complete` 可以获取协程返回值，若没有，则返回 None**

#### 协程调用协程 ####

一个协程可以启动另一个协程，从而可以任务根据工作内容，封装到不同的协程中。我们可以在协程中使用**await**关键字，链式的调度协程，来形成一个协程任务流：

import asyncio
    
    
    async def foo():
        print('开始运行协程，主协程')
        print('等待 result1 协程运行')
        res1 = await result1()
        print('等待 result2 协程运行')
        res2 = await result2('rose')
        return res1, res2
    
    
    async def result1():
        print('result1 协程')
        return 'result1'
    
    
    async def result2(name):
        print('result2 协程')
        return 'result2 协程接收了一个参数：', name
    
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        try:
            result = loop.run_until_complete(foo())
            print('返回值：', result)
        finally:
            print('关闭事件循环')
            loop.close()

运行结果：

开始运行协程，主协程
    等待 result1 协程运行
    result1 协程
    等待 result2 协程运行
    result2 协程
    返回值： ('result1', ('result2 协程接收了一个参数：', 'rose'))
    关闭事件循环

#### 协程中调用普通函数 ####

在协程中可以通过一些方法去调用普通的函数。可以使用的关键字有`call_soon、call_later、call_at`

**call\_soon**

调用立即返回

loop.call_soon(callback, *args, context=None)

大部分的回调函数支持位置参数，而**不支持”关键字参数”**，如果是想要使用关键字参数，则推荐使用`functools.aprtial()`对方法进一步包装。可选关键字context允许指定要运行的回调的自定义`contextvars.Context`。当没有提供上下文时使用当前上下文。在Python 3.7中， `asyncio`  
协程加入了对上下文的支持。使用上下文就可以在一些场景下隐式地传递变量，比如数据库连接`session`等，而不需要在所有方法调用显示地传递这些变量。

import asyncio
    import functools
    
    
    def callback(args, *, kwargs='default'):
        print('普通函数 callback作为回调函数，获取参数：', args, kwargs)
    
    
    async def main(loop):
        print('注册 callback')
        loop.call_soon(callback, 1)
        loop.call_soon(callback, kwagrs='rose')
        # wrapped = functools.partial(callback, kwagrs='not default')
        # loop.call_soon(wrapped, 2)
        await asyncio.sleep(0.2)
    
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        try:
            loop.run_until_complete(main(loop))
        finally:
            loop.close()

运行结果：

注册 callback
    Traceback (most recent call last):
    普通函数 callback作为回调函数，获取参数： 1 default
      File "D:/pycharm resource/Projects/TestDeploy/协程/协程调用普通函数.py", line 21, in <module>
        loop.run_until_complete(main(loop))
      File "C:\Python35\Lib\asyncio\base_events.py", line 342, in run_until_complete
        return future.result()
      File "C:\Python35\Lib\asyncio\futures.py", line 274, in result
        raise self._exception
      File "C:\Python35\Lib\asyncio\tasks.py", line 239, in _step
        result = coro.send(value)
      File "D:/pycharm resource/Projects/TestDeploy/协程/协程调用普通函数.py", line 12, in main
        loop.call_soon(callback, kwagrs='rose')
    TypeError: call_soon() got an unexpected keyword argument 'kwagrs'

这里用的是 `Python3.5`，所以 `asyncio` 没有对上下文的支持

--------------------

**call\_later**

延时调用一个函数

loop.call_later(delay, callback, *args, context=None)		# 事件循环在delay多长时间之后才执行callback函数

import asyncio
    from time import ctime
    
    
    def callback(n):
        print('普通函数 callback作为回调函数，获取参数：', n)
    
    
    async def main(loop):
        print('注册 callback')
        print('起始时间', ctime())
        loop.call_later(5, callback, 1)
        print('第一次延迟调用：', ctime())
        loop.call_later(10, callback, 2)
        print('第二次延迟调用：', ctime())
        loop.call_soon(callback, 3)
    
        await asyncio.sleep(2)
    
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        try:
            loop.run_until_complete(main(loop))
        finally:
            print('协程结束', ctime())
            loop.close()

运行结果：

注册 callback
    起始时间 Sat Oct 12 10:28:29 2019
    第一次延迟调用： Sat Oct 12 10:28:29 2019
    第二次延迟调用： Sat Oct 12 10:28:29 2019
    普通函数 callback作为回调函数，获取参数： 3
    协程结束 Sat Oct 12 10:28:31 2019

总结：

*  **call\_soon**会在 `call_later` 之前执行，和它的位置在哪无关
 *  **call\_later**的第一个参数越小，越先执行。

--------------------

**call\_at**

loop.call_at(when, callback, *args, context=None)

第一个参数的含义代表的是一个单调时间，它和我们平时说的系统时间有点差异，指的是事件循环内部时间，可以通过`loop.time()`获取，然后可以在此基础上进行操作。`call_later` 内部实质是调用 `call_at`：

import asyncio
    
    
    def call_back(n, loop):
        print("callback %s  运行时间点 %s" % (n, loop.time()))
    
    
    async def main(loop):
        now = loop.time()
        print("当前的内部时间", now)
        print("循环时间", now)
        print("注册callback")
        loop.call_at(now + 0.1, call_back, 1, loop)
        loop.call_at(now + 0.2, call_back, 2, loop)
        loop.call_soon(call_back, 3, loop)
        await asyncio.sleep(1)
    
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        try:
            print("进入事件循环")
            loop.run_until_complete(main(loop))
        finally:
            print("关闭循环")
            loop.close()

运行结果：

进入事件循环
    当前的内部时间 148978.593
    循环时间 148978.593
    注册callback
    callback 3  运行时间点 148978.593
    callback 1  运行时间点 148978.703
    callback 2  运行时间点 148978.796
    关闭循环