Python实现线程的高效非阻塞I/O调用

off999 2024-11-25 15:54 33 浏览 0 评论

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

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

协程的一般使用方法

import asyncio

async def do_some_work(x):
    print('Waiting: ', x)
    # 协程遇到await，事件循环将会挂起该协程，执行别的协程，直到其他的协程也挂起或者执行完毕，再进行下一个协程的执行。
    # 此处先挂起，再执行await的协程,最后执行return
    await 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 time

async 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 ClientSession

url = "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 aiomysql

async 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 == 42
    pool.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.ioloop

class 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 uvloop

class 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方式时间非阻塞了。