Python 异步 IO(asyncio)、多进程、多线程性能对

作者: 刀尖红叶 | 来源:发表于2019-11-22 13:17 被阅读0次

Python 异步 IO(asyncio)、多进程、多线程性能对
快乐的 asyncio 学习
Python 异步IO - asyncio
python asyncio构建服务器
并发
asyncio
利用aiohttp制作异步爬虫
利用aiohttp制作异步爬虫
python中的异步操作
Python 经验 - 异步编程与 Asyncio

IO 密集型应用
IO 密集型应用CPU等待IO时间远大于CPU 自身运行时间，太浪费；常见的 IO 密集型业务包括：浏览器交互、磁盘请求、网络爬虫、数据库请求等

image.png

Python 世界对于 IO 密集型场景的并发提升有 3 种方法：多进程、多线程、异步 IO(asyncio);理论上讲asyncio是性能最高的，原因如下：
1.进程、线程会有CPU上下文切换
2.进程、线程需要内核态和用户态的交互，性能开销大；而协程对内核透明的,只在用户态运行
3.进程、线程并不可以无限创建，最佳实践一般是 CPU*2；而协程并发能力强，并发上限理论上取决于操作系统IO多路复用(Linux下是 epoll)可注册的文件描述符的极限

那asyncio的实际表现是否如理论上那么强，到底强多少呢？我构建了如下测试场景：

访问500台 DB，并sleep 100ms模拟业务查询 
方法 1；顺序串行一台台执行
方法 2：多进程
方法 3：多线程
方法 4：asyncio
方法 5：asyncio+uvloop

最后的asyncio+uvloop和官方asyncio 最大不同是用 Cython+libuv 重新实现了asyncio 的事件循环(event loop)部分,官方测试性能是 node.js的 2 倍，持平 golang。

以下测试代码需要 Pyhton3.7+：
顺序串行一台台执行

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import records

user=xx
pass=xx
port=xx
hosts= [....] # 500台 db列表

def query(host):
    conn = records.Database(
    f'mysql+pymysql://{user}:{pass}@{host}:{port}/mysql?charset=utf8mb4')
    rows = conn.query('select sleep(0.1);')
    print(rows[0])

def main():
    for h in hosts:
        query(h)

# main entrance
if __name__ == '__main__':
    main()

多进程

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from concurrent import futures
import records

user=xx
pass=xx
port=xx
hosts= [....] # 500台 db列表

def query(host):
    conn = records.Database(
    f'mysql+pymysql://{user}:{pass}@{host}:{port}/mysql?charset=utf8mb4')
    rows = conn.query('select sleep(0.1);')
    print(rows[0])

def main():
    with futures.ProcessPoolExecutor() as executor:
        for future in executor.map(query,hosts):
            pass

# main entrance
if __name__ == '__main__':
    main()

多线程

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from concurrent import futures
import records

user=xx
pass=xx
port=xx
hosts= [....] # 500台 db列表

def query(host):
    conn = records.Database(
    f'mysql+pymysql://{user}:{pass}@{host}:{port}/mysql?charset=utf8mb4')
    rows = conn.query('select sleep(0.1);')
    print(rows[0])

def main():
    with futures.ThreadPoolExecutor() as executor:
        for future in executor.map(query,hosts):
            pass

# main entrance
if __name__ == '__main__':
    main()

asyncio

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
from databases import Database 

user=xx
pass=xx
port=xx
hosts= [....] # 500台 db列表

async def query(host):
    DATABASE_URL = f'mysql+pymysql://{user}:{pass}@{host}:{port}/mysql?charset=utf8mb4'
    async with Database(DATABASE_URL) as database:
        query = 'select sleep(0.1);'
        rows = await database.fetch_all(query=query)
        print(rows[0])

async def main():
    tasks = [asyncio.create_task(query(host)) for host in hosts]
    await asyncio.gather(*tasks)

# main entrance
if __name__ == '__main__':
    asyncio.run(main())

asyncio+uvloop

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import uvloop
from databases import Database 

user=xx
pass=xx
port=xx
hosts= [....] # 500台 db列表

async def query(host):
    DATABASE_URL = f'mysql+pymysql://{user}:{pass}@{host}:{port}/mysql?charset=utf8mb4'
    async with Database(DATABASE_URL) as database:
        query = 'select sleep(0.1);'
        rows = await database.fetch_all(query=query)
        print(rows[0])

async def main():
    tasks = [asyncio.create_task(query(host)) for host in hosts]
    await asyncio.gather(*tasks)

# main entrance
if __name__ == '__main__':
    uvloop.install()
    asyncio.run(main())

运行时间对比