python 常用的异步框架汇总整理_Python

正文开始

asyncio 是 Python 3.4版本引入的标准库，直接内置了对异步IO的支持。

asyncio 在单线程内部维护了 EventLoop 队列，然后把需要执行异步IO的任务添加到 EventLoop 队列中，至于任务的完成通过类似回调的逻辑是实现后续的任务。如果你有 JavaScript的基础那么理解python的 asyncio 很简单，关键字、语法以及实现的原理都极其类似。

				?

									import asyncio

									async def main():

									    print('Hello ...')

									    await asyncio.sleep(1)

									    print('... World!')

									# Python 3.7+

									asyncio.run(main())

1. Tornado

Tornado 是一个Python web框架和异步网络库，起初由 FriendFeed 开发. 通过使用非阻塞网络I/O， Tornado可以支撑上万级的连接，处理长连接, WebSockets ，和其他需要与每个用户保持长久连接的应用。

下面贴上官方 demo ：

				?

									import tornado.ioloop

									import tornado.web

									class MainHandler(tornado.web.RequestHandler):

									    def get(self):

									        self.write("Hello, world")

									def make_app():

									    return tornado.web.Application([

									        (r"/", MainHandler),

									    ])

									if __name__ == "__main__":

									    app = make_app()

									    app.listen(8888)

									    tornado.ioloop.IOLoop.current().start()

2. Aiohttp

一个基于 asyncio 异步的web框架，支持 websocket,不需要写回掉的代码、有着丰富的生态、中间价等、开箱即用的服务端与客户端。

下面贴上官方 demo ：

				?

									# 客服端代码

									import aiohttp

									import asyncio

									async def main():

									    async with aiohttp.ClientSession() as session:

									        async with session.get('http://python.org') as response:

									            print("Status:", response.status)

									            print("Content-type:", response.headers['content-type'])

									            html = await response.text()

									            print("Body:", html[:15], "...")

									loop = asyncio.get_event_loop()

									loop.run_until_complete(main())

				?

									# 服务端代码

									from aiohttp import web

									async def handle(request):

									    name = request.match_info.get('name', "Anonymous")

									    text = "Hello, " + name

									    return web.Response(text=text)

									async def wshandle(request):

									    ws = web.WebSocketResponse()

									    await ws.prepare(request)

									    async for msg in ws:

									        if msg.type == web.WSMsgType.text:

									            await ws.send_str("Hello, {}".format(msg.data))

									        elif msg.type == web.WSMsgType.binary:

									            await ws.send_bytes(msg.data)

									        elif msg.type == web.WSMsgType.close:

									            break

									    return ws

									app = web.Application()

									app.add_routes([web.get('/', handle),

									                web.get('/echo', wshandle),

									                web.get('/{name}', handle)])

									if __name__ == '__main__':

									    web.run_app(app)

aiohttp的生态：

aiohttp-session 为 aiohttp 服务提供 sessions支持，同时支持数据持久化数据库。
aiohttp-debugtoolbar 为 aiohttp 提供调试的工具（记录 asyncio 异常的堆栈信息）。
aiohttp-security 为aiohttp提供认证以及权限相关的插件。
aiohttp-devtools aiohttp开发工具，提供开发环境的部署、静态资源的代理。
aiohttp-cors CORS 跨域认证支持。
aiohttp-sse 服务端事件的支持（一种服务端推送消息的服务）。
pytest-aiohttp pytest 测试框架的支持。
aiohttp-mako Mako 服务端模板渲染的支持。
aiohttp-jinja2 Jinja2 服务端模板渲染的支持（大名鼎鼎的flask的渲染引擎）。
aiozipkin 分布式系统中对服务追踪，为微服务中延时问题提供数据支持。

aiohttp数据库支持：

aiopg PostgreSQL异步支持。
aiomysql MySql 异步支持。
aioredis Redis 异步支持。
asyncpg 另外一个对 PostgreSQL 异步支持，比 aiopg 效率高，但是 api 不通用。

3.Sanic

Sanic 是一个 Python 3.7+ 的基于 asyncio 的 web 服务器和web框架，目标是提供一种简单的方法来启动和运行一个易于构建、扩展和终极性能HTTP服务器，是一个比较类似 falsk 的异步web框架。

To provide a simple way to get up and running a highly performant HTTP server that is easy to build, to expand, and ultimately to scale.

官方demo：

				?

									from sanic import Sanic

									from sanic.response import json

									app = Sanic("My Hello, world app")

									@app.route('/')

									async def test(request):

									    return json({'hello': 'world'})

									if __name__ == '__main__':

									    app.run()

4. FastAPI

FastAPI 是一个用于构建API的高性能web框架，基于Python3.6+并支持标准的 Python 类型提示。同时是最快的 Python web框架之一，可与 NodeJS 和 Go 比肩（主要归功于 Starlette 和 Pydantic）。

				?

									from typing import Optional

									from fastapi import FastAPI

									app = FastAPI()

									@app.get("/")

									def read_root():

									    return {"Hello": "World"}

									@app.get("/items/{item_id}")

									def read_item(item_id: int, q: Optional[str] = None):

									    return {"item_id": item_id, "q": q}

									# 启动 uvicorn main:app --reload

									# pip install uvicorn[standard]

5. Ruia

一个基于asyncio和aiohttp的异步爬虫框架，目标在于让开发者编写爬虫尽可能地方便快速。国人开发中文文档的支持，方便快速的构建爬虫项目，自定义HTML解析工具，快速获取页面数据。

官方demo：

				?

									import asyncio

									from ruia import Item, TextField, AttrField

									class HackerNewsItem(Item):

									    target_item = TextField(css_select='tr.athing')

									    title = TextField(css_select='a.storylink')

									    url = AttrField(css_select='a.storylink', attr='href')

									async def test_item():

									    url = 'https://news.ycombinator.com/news?p=1'

									    async for item in HackerNewsItem.get_items(url=url):

									        print('{}: {}'.format(item.title, item.url))

									if __name__ == '__main__':

									    # Python 3.7 Required.

									    asyncio.run(test_item()) 

									    # For Python 3.6

									    # loop = asyncio.get_event_loop()

									    # loop.run_until_complete(test_item())