Python基于async和await实现非阻塞I/O调用

• 前言
• python协程与I/O调用的现状
• • 协程的一般使用方法
  • Python中一般的I/O调用方法
• 常用的异步I/O库
• • • aiohttp库
    • aiomysql库
• 关于async和await
• 结合Tornado框架实现异步服务器
• • 基本的实现方式
  • 非阻塞的实现方式
  • Tornado部署最佳实践
• 参考文献

前言

前面的章节中讲述过Python的协程及其使用方法，在众多网络模型中，为了实现高并发也有很多方案，多线程，多进程和协程。而协程的方式，调度来自用户，用户可以在函数中yield一个状态，实现程序的非阻塞使用。使用协程可以实现高效的并发任务。Python3.5之后出现的async/await的使用方法，本文将详细讲述async/await的使用以及结合Tornado实现非阻塞服务器的方法。

python协程与I/O调用的现状

协程的一般使用方法

import asyncioasync def do_some_work(x):print('Waiting: ', x)# 协程遇到await，事件循环将会挂起该协程，执行别的协程，直到其他的协程也挂起或者执行完毕，再进行下一个协程的执行。# 此处先挂起，再执行await的协程,最后执行returnawait asyncio.sleep(x)return 'Done after {}s'.format(x)coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)

上面的例子中，实现了协程，如果有多个coroutine同时执行do_some_work函数，可以达到非阻塞的效果，而不必阻塞等待每一个sleep完成

但是在一般python开发中，我们会遇到很多的I/O调用，如网络I/O请求、数据库I/O请求等，这些调用执行时间都比较长，如果有很多请求都要进行I/O操作，那么Python的单个进程模型里面，I/O会阻塞很多时间，导致后面的请求响应时间过长甚至超时，所以我们需要将更多的耗时I/O操作都采用非阻塞的方式，才能大大提高系统的执行效率和性能。

Python中一般的I/O调用方法

一般情况下，我们Python在实现I/O调用时，例如http请求，都会用requests库或者自带的urllib库来实现，方法如下：

from urllib import parse,request
import json# POST请求 当request中包含data参数的时候,是POST请求,反之是GET请求
textmod = {"username": "admin", "password": "123456"}
textmod = json.dumps(textmod).encode(encoding='utf-8')
header_dict = {'Accept': 'application/json', 'Content-Type': 'application/json'}
url = 'http://localhost:8080/api/xxx'
req = request.Request(url=url, data=textmod, headers=header_dict)
res = request.urlopen(req)
res = res.read()

上面这种一般的I/O调用方法就是阻塞式的，常用的还是连接数据库的SQLAlchemy、Django的orm等等。如果采用这种方式部署在我们的api服务器,如flask、django等，就会在一个线程内造成阻塞，必须要采用多进程如gevent库、多线程模型等才能扛得起并发。

下面我们根据常用的I/O调用库来介绍一些实现了aio(fei阻塞)的I/O调用库。

常用的异步I/O库

假如我们用协程实现一个sleep：

import asyncio
import timeasync def do_some_work(x):print('Waiting: ', x)time.sleep(x)return 'Done after {}s'.format(x)coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)

这种方式虽然定义了async方法，但是这是无法实现非阻塞的，程序运行到time.sleep时还是会阻塞x秒，无法使用await实现非阻塞，所以我们必须把time.sleep改为await asyncio.sleep(x)才能达到非阻塞的效果。
同样的，像urllib、requests、SQLAlchemy等都库的执行，都是阻塞式的，一个线程里面运行到对应的I/O调用方法时总会等待执行返回，而解决这种办法的途径只有多线程和协程，多线程在Python里面由于GIL锁的存在，效率有限而且不好管理，所以还是推荐采用协程的方法解决。这里介绍一下几个实现了aio的http、mysql库，来源于第三方开发者，详细的实现方法可以在github上查看。

aiohttp库

import asyncio
from aiohttp import ClientSessionurl = "https://www.baidu.com/{}"
async def hello(url):async with ClientSession() as session:async with session.get(url) as response:response = await response.read()print(response)if __name__ == '__main__':coroutine = hello(url)loop = asyncio.get_event_loop()loop.run_until_complete(coroutine)

上面的例子就可以实现非阻塞的http请求，当多个coroutine任务一起放入loop.run_until_complete()方法执行时，coroutine1运行到await response.read()的时候，进入等待，但是同时coroutine2可以开始执行而不需要等待coroutine1，从而实现了非阻塞。
同样的方式还有aiomysql、aioredis等等

aiomysql库

import asyncio
import aiomysqlasync def test_example(loop):pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,user='root', password='',db='mysql', loop=loop)async with pool.acquire() as conn:async with conn.cursor() as cur:await cur.execute("SELECT 42;")print(cur.description)(r,) = await cur.fetchone()assert r == 42pool.close()await pool.wait_closed()loop = asyncio.get_event_loop()
loop.run_until_complete(test_example(loop))

采用aiomysql库，创建数据库表、执行CRUD操作时都可以非阻塞，毕竟mysql本身是支持多线程的。

关于async和await

python里面async和await到底是怎么实现的呢？
其实在旧版本的python里面是采用yield关键字，将方法封装成一个生成器，让每个要实现协程的方法通过装饰器函数@coroutine的方式来封装的。async和await是3.5版本之后的新语法，但是实现方式大同小异。
例如上面实现的asyncio.sleep()方法，在coroutine1、coroutine2等多个任务都传入loop.run_until_complete()时，程序执行到coroutine1的sleep会作为生成器变量保存起来，然后挂起coroutine1，执行coroutine2，而整个loop.run_until_complete()方式会遍历生成器的值，最后得到每个coroutine的结果再分别执行后面步骤，从而实现了非阻塞的效果。
但是该方式还是无法利用到多核CPU，所以现在服务器的最佳部署方式还是多进程+协程的方式

结合Tornado框架实现异步服务器

说完了python以脚本方式直接运行的协程实现方法，我们要来看看如果是服务器该如何实现协程方式运作。
python常用的flask、Django等框架，在运行api服务，监听端口时，是无法实现非阻塞的，所以flask和Django的部署方式常常采用了多线程来提高并发效率。
近几年出现的Tornado框架是采用了I/O多路复用的epoll机制来实现的，这是称作一种uvloop。底层方面，从一开始的loop封装慢慢发展为基于python3.4之后出现的asyncio库来封装，所以Tornado框架是由事件驱动来作为api服务的。

基本的实现方式

import tornado.web
import tornado.httpserver
import tornado.ioloopclass IndexPageHandler(tornado.web.RequestHandler):def get(self):self.render('tornado_index.html')class Application(tornado.web.Application):def __init__(self):handlers = [(r'/', IndexPageHandler),]settings = {"template_path": "templates/"}tornado.web.Application.__init__(self, handlers, **settings)if __name__ == '__main__':app = Application()server = tornado.httpserver.HTTPServer(app)server.listen(5000)tornado.ioloop.IOLoop.instance().start()

上面tornado框架的启动方式，是采用之前的ioloop方式的，这种方式如果在我们的控制器IndexPageHandler里面定义了async get方法，则里面的await 是不能实现非阻塞的。所以我们要采用为以下的asyncio+uvloop的事件循环机制才能实现非阻塞：

非阻塞的实现方式

import tornado.web
import tornado.httpserver
import tornado.ioloop
import tornado.platform.asyncio as tornado_asyncio
import asyncio
import uvloopclass IndexPageHandler(tornado.web.RequestHandler):def get(self):self.render('tornado_index.html')class Application(tornado.web.Application):def __init__(self):handlers = [(r'/', IndexPageHandler),]settings = {"template_path": "templates/"}tornado.web.Application.__init__(self, handlers, **settings)if __name__ == '__main__':asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())tornado_asyncio.AsyncIOMainLoop().install()app = Application()server = tornado.httpserver.HTTPServer(app)server.listen(5000)asyncio.get_event_loop().run_forever()

采用这种方式启动Tornado，就可以在我们的IndexPageHandler等多个Handler里面定义async方法，然后引入我们的aio库，用await方式时间非阻塞了。

Tornado部署最佳实践

1. 采用非阻塞方式驱动Tornado框架；
2. 采用多进程方式运行api服务；
3. 如果需要进行环境隔离或者快速扩容等部署，建议采用Docker方式实现

参考文献

1. https://blog.csdn.net/brucewong0516/article/details/82697935
2. https://www.jianshu.com/p/b5e347b3a17c
3. https://juejin.im/post/6844903832716247053

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！