Python asyncio高性能异步编程 详解

11秒前 126阅读

目录

一、协程

1.1、greenlet实现协程

1.2、yield关键字

1.3、asyncio

1.4、async & await关键字

二、协程意义

三、异步编程

3.1、事件循环

3.2、快速上手

3.3、await

3.4、Task对象

3.5、asyncio.Future对象

3.5、concurrent.futures.Future对象

3.7、异步迭代器

3.8、异步上下文管理器

四、uvloop

五、实战案例

5.1、异步redis

5.2、异步MySQL

5.3、FastAPI框架

六、总结


一、协程

协程不是计算机提供,程序员人为创造。

协程(Coroutine),也可以被称为微线程,是一种用户态内的上下文切换技术。简而言之,其实就是通过一个线程实现代码块相互切换执行。例如:

def func1():
	print(1)
    ...
	print(2)
	
def func2():
	print(3)
    ...
	print(4)
func1()
func2()

实现协程有这么几种方法:

  • greenlet,早期模块。

  • yield关键字。

  • asyncio装饰器(py3.4)

  • async、await关键字(py3.5)【推荐】

    1.1、greenlet实现协程

    pip3 install greenlet
    from greenlet import greenlet
    def func1():
        print(1)        # 第2步:输出 1
        gr2.switch()    # 第3步:切换到 func2 函数
        print(2)        # 第6步:输出 2
        gr2.switch()    # 第7步:切换到 func2 函数,从上一次执行的位置继续向后执行
    def func2():
        print(3)        # 第4步:输出 3
        gr1.switch()    # 第5步:切换到 func1 函数,从上一次执行的位置继续向后执行
        print(4)        # 第8步:输出 4
    gr1 = greenlet(func1)
    gr2 = greenlet(func2)
    gr1.switch() # 第1步:去执行 func1 函数

    Python asyncio高性能异步编程 详解

    1.2、yield关键字

    def func1():
        yield 1
        yield from func2()
        yield 2
    def func2():
        yield 3
        yield 4
    f1 = func1()
    for item in f1:
        print(item)

    Python asyncio高性能异步编程 详解

    1.3、asyncio

    在python3.4及之后的版本。

    import asyncio
    @asyncio.coroutine
    def func1():
        print(1)
        # 网络IO请求:下载一张图片
        yield from asyncio.sleep(2)  # 遇到IO耗时操作,自动化切换到tasks中的其他任务
        print(2)
    @asyncio.coroutine
    def func2():
        print(3)
        # 网络IO请求:下载一张图片
        yield from asyncio.sleep(2) # 遇到IO耗时操作,自动化切换到tasks中的其他任务
        print(4)
    tasks = [
        asyncio.ensure_future(func1()),
        asyncio.ensure_future(func2())
    ]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

    Python asyncio高性能异步编程 详解

    我们将两个函数放到tasks中,启动时会随机选一个函数执行,遇到IO阻塞自动切换。

    1.4、async & await关键字

    在python3.5及之后的版本。

    import asyncio
    async def func1():
        print(1)
        # 网络IO请求:下载一张图片
        await asyncio.sleep(2)  # 遇到IO耗时操作,自动化切换到tasks中的其他任务
        print(2)
    async def func2():
        print(3)
        # 网络IO请求:下载一张图片
        await asyncio.sleep(2) # 遇到IO耗时操作,自动化切换到tasks中的其他任务
        print(4)
    tasks = [
        asyncio.ensure_future( func1() ),
        asyncio.ensure_future( func2() )
    ]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

    Python asyncio高性能异步编程 详解

    二、协程意义

    在一个线程中如果遇到IO等待时间,线程不会傻傻等,利用空闲的时候再去干点其他事。

    案例:去下载三张图片(网络IO)。

    (1)普通方式(同步)

    pip install requests
    import requests
    def download_image(url):
        print("开始下载:",url)
        # 发送网络请求,下载图片
        response = requests.get(url)
        print("下载完成")
        # 图片保存到本地文件
        file_name = url.rsplit('_')[-1]
        with open(file_name, mode='wb') as file_object:
            file_object.write(response.content)
    if __name__ == '__main__':
        url_list = [
            'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
            'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
            'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
        ]
        for item in url_list:
            download_image(item)

    Python asyncio高性能异步编程 详解

    (2)协程方式(异步)

    pip install aiohttp
    import aiohttp
    import asyncio
    INSTALL_AIOHTTP = """pip3 install aiohttp"""
    async def fetch(session, url):
        print("发送请求:", url)
        async with session.get(url, verify_ssl=False) as response:
            content = await response.content.read()
            file_name = url.rsplit('_')[-1]
            with open(file_name, mode='wb') as file_object:
                file_object.write(content)
            print('下载完成', url)
    async def main():
        async with aiohttp.ClientSession() as session:
            url_list = [
                'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
                'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
                'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
            ]
            tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
            await asyncio.wait(tasks)
    if __name__ == '__main__':
        asyncio.run(main())

    Python asyncio高性能异步编程 详解

    三、异步编程

    3.1、事件循环

    理解成为一个死循环 ,去检测并执行某些代码。

    # 伪代码
    任务列表 = [ 任务1, 任务2, 任务3,... ]
    while True:
        可执行的任务列表,已完成的任务列表 = 去任务列表中检查所有的任务,将'可执行'和'已完成'的任务返回
        
        for 就绪任务 in 可执行的任务列表:
            执行已就绪的任务
            
        for 已完成的任务 in 已完成的任务列表:
            在任务列表中移除 已完成的任务
    	如果 任务列表 中的任务都已完成,则终止循环
    import asyncio
    # 去生成或获取一个事件循环
    loop = asyncio.get_event_loop()
    # 将任务放到`任务列表`
    loop.run_until_complete(任务)

    3.2、快速上手

    协程函数,定义函数时候 async def 函数名 。

    协程对象,执行 协程函数() 得到的协程对象。

    async def func():
        pass
    result = func() # 协程对象

    注意:执行协程函数创建协程对象,函数内部代码不会执行。

    如果想要运行协程函数内部代码,必须要将协程对象交给事件循环来处理。

    import asyncio 
    async def func():
        print("快来搞我吧!")
    result = func()
    # loop = asyncio.get_event_loop()
    # loop.run_until_complete( result )
    asyncio.run(result) # python3.7 

    3.3、await

    await + 可等待的对象(协程对象、Future、Task对象 -> IO等待)

    示例1:

    import asyncio
    async def func():
        print("来玩呀")
        response = await asyncio.sleep(2)
        print("结束", response)
    asyncio.run(func())

    示例2:

    import asyncio
    async def others():
        print("start")
        await asyncio.sleep(2)
        print('end')
        return '返回值'
    async def func():
        print("执行协程函数内部代码")
        # 遇到IO操作挂起当前协程(任务),等IO操作完成之后再继续往下执行。当前协程挂起时,事件循环可以去执行其他协程(任务)。
        response = await others()
        print("IO请求结束,结果为:", response)
        
    asyncio.run(func())

    Python asyncio高性能异步编程 详解

    示例3:

    import asyncio
    async def others():
        print("start")
        await asyncio.sleep(2)
        print('end')
        return '返回值'
    async def func():
        print("执行协程函数内部代码")
        # 遇到IO操作挂起当前协程(任务),等IO操作完成之后再继续往下执行。当前协程挂起时,事件循环可以去执行其他协程(任务)。
        response1 = await others()
        print("IO请求结束,结果为:", response1)
        
        response2 = await others()
        print("IO请求结束,结果为:", response2)
        
    asyncio.run(func())

    Python asyncio高性能异步编程 详解

    await就是等待对象的值得到结果之后再继续向下走,其实就是同步操作。

    3.4、Task对象

    Tasks用于并发调度协程,通过asyncio.create_task(协程对象)的方式创建Task对象,这样可以让协程加入事件循环中等待被调度执行。除了使用 asyncio.create_task() 函数以外,还可以用低层级的 loop.create_task() 或 ensure_future() 函数。不建议手动实例化 Task 对象。

    注意:asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.7 之前,可以改用低层级的 asyncio.ensure_future() 函数。

    示例1:

    import asyncio
    async def func():
        print(1)
        await asyncio.sleep(2)
        print(2)
        return "返回值"
    async def main():
        print("main开始")
        # 创建Task对象,将当前执行func函数任务添加到事件循环。
        task1 = asyncio.create_task(func())
        # 创建Task对象,将当前执行func函数任务添加到事件循环。
        task2 = asyncio.create_task(func())
        print("main结束")
        # 当执行某协程遇到IO操作时,会自动化切换执行其他任务。
        # 此处的await是等待相对应的协程全都执行完毕并获取结果
        ret1 = await task1
        ret2 = await task2
        print(ret1, ret2)
    asyncio.run(main())

    Python asyncio高性能异步编程 详解

    示例2:

    import asyncio
    async def func():
        print(1)
        await asyncio.sleep(2)
        print(2)
        return "返回值"
    async def main():
        print("main开始")
        task_list = [
            asyncio.create_task(func(), name='n1'),
            asyncio.create_task(func(), name='n2')
        ]
        print("main结束")
        done, pending = await asyncio.wait(task_list, timeout=None)
        print(done)
    asyncio.run(main())

    Python asyncio高性能异步编程 详解

    示例3:

    import asyncio
    async def func():
        print(1)
        await asyncio.sleep(2)
        print(2)
        return "返回值"
    task_list = [
        func(),
    	func(), 
    ]
    done,pending = asyncio.run(asyncio.wait(task_list))
    print(done)

    3.5、asyncio.Future对象

    Task继承Future,Task对象内部await结果的处理基于Future对象来的。

    示例1:

    import asyncio
    async def main():
        # 获取当前事件循环
        loop = asyncio.get_running_loop()
        # 创建一个任务(Future对象),这个任务什么都不干。
        fut = loop.create_future()
        # 等待任务最终结果(Future对象),没有结果则会一直等下去。
        await fut
    asyncio.run(main())

    示例2:

    import asyncio
    async def set_after(fut):
        await asyncio.sleep(2)
        fut.set_result("666")
    async def main():
        # 获取当前事件循环
        loop = asyncio.get_running_loop()
        # 创建一个任务(Future对象),没绑定任何行为,则这个任务永远不知道什么时候结束。
        fut = loop.create_future()
        # 创建一个任务(Task对象),绑定了set_after函数,函数内部在2s之后,会给fut赋值。
        # 即手动设置future任务的最终结果,那么fut就可以结束了。
        await loop.create_task(  set_after(fut) )
        # 等待 Future对象获取 最终结果,否则一直等下去
        data = await fut
        print(data)
    asyncio.run(main())

    3.5、concurrent.futures.Future对象

    使用线程池、进程池实现异步操作时用到的对象。

    import time
    from concurrent.futures import Future
    from concurrent.futures.thread import ThreadPoolExecutor
    from concurrent.futures.process import ProcessPoolExecutor
    def func(value):
        time.sleep(1)
        print(value)
        return 123
    # 创建线程池
    pool = ThreadPoolExecutor(max_workers=5)
    # 创建进程池
    # pool = ProcessPoolExecutor(max_workers=5)
    for i in range(10):
        fut = pool.submit(func, i)
        print(fut)

    案例:asyncio + 不支持异步的模块

    import asyncio
    import requests
    async def download_image(url):
        # 发送网络请求,下载图片(遇到网络下载图片的IO请求,自动化切换到其他任务)
        print("开始下载:", url)
        loop = asyncio.get_event_loop()
        # requests模块默认不支持异步操作,所以就使用线程池来配合实现了。
        future = loop.run_in_executor(None, requests.get, url)
        response = await future
        print('下载完成')
        # 图片保存到本地文件
        file_name = url.rsplit('_')[-1]
        with open(file_name, mode='wb') as file_object:
            file_object.write(response.content)
    if __name__ == '__main__':
        url_list = [
            'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
            'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
            'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
        ]
        tasks = [download_image(url) for url in url_list]
        loop = asyncio.get_event_loop()
        loop.run_until_complete(asyncio.wait(tasks))

    Python asyncio高性能异步编程 详解

    3.7、异步迭代器

    什么是异步迭代器

    实现了 __aiter__() 和 __anext__() 方法的对象。__anext__ 必须返回一个 awaitable 对象。async for 会处理异步迭代器的 __anext__() 方法所返回的可等待对象,直到其引发一个 StopAsyncIteration 异常。由 PEP 492 引入。

    什么是异步可迭代对象?

    可在 async for 语句中被使用的对象。必须通过它的 __aiter__() 方法返回一个 asynchronous iterator。由 PEP 492 引入。

    import asyncio
    class Reader(object):
        """ 自定义异步迭代器(同时也是异步可迭代对象) """
        def __init__(self):
            self.count = 0
        async def readline(self):
            # await asyncio.sleep(1)
            self.count += 1
            if self.count == 100:
                return None
            return self.count
        def __aiter__(self):
            return self
        async def __anext__(self):
            val = await self.readline()
            if val == None:
                raise StopAsyncIteration
            return val
        
    async def func():
        obj = Reader()
        async for item in obj:
            print(item)
            
    asyncio.run( func() )

    3.8、异步上下文管理器

    此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。

    import asyncio
    class AsyncContextManager:
    	def __init__(self):
            self.conn = conn
            
        async def do_something(self):
            # 异步操作数据库
            return 666
        async def __aenter__(self):
            # 异步链接数据库
            self.conn = await asyncio.sleep(1)
            return self
        async def __aexit__(self, exc_type, exc, tb):
            # 异步关闭数据库链接
    		await asyncio.sleep(1)
    async def func():
        async with AsyncContextManager() as f:
            result = await f.do_something()
            print(result)
    asyncio.run( func() )

    四、uvloop

    是asyncio的事件循环的替代方案。事件循环 > 默认asyncio的事件循环。

    pip install uvloop
    import asyncio
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    # 编写asyncio的代码,与之前写的代码一致。
    # 内部的事件循环自动化会变为uvloop
    asyncio.run(...)

    注意:一个asgi -> uvicorn 内部使用的就是uvloop

    五、实战案例

    5.1、异步redis

    在使用python代码操作redis时,链接/操作/断开都是网络IO。

    案例1:

    pip install aioredis
    import asyncio
    import aioredis
    async def execute(address, password):
        print("开始执行", address)
        # 网络IO操作:创建redis连接
        redis = await aioredis.create_redis(address, password=password)
        # 网络IO操作:在redis中设置哈希值car,内部在设三个键值对,即: redis = { car:{key1:1,key2:2,key3:3}}
        await redis.hmset_dict('car', key1=1, key2=2, key3=3)
        # 网络IO操作:去redis中获取值
        result = await redis.hgetall('car', encoding='utf-8')
        print(result)
        redis.close()
        # 网络IO操作:关闭redis连接
        await redis.wait_closed()
        print("结束", address)
    asyncio.run(execute('redis://127.0.0.1:6379', "123456"))

    示例2:

    import asyncio
    import aioredis
    async def execute(address, password):
        print("开始执行", address)
        # 网络IO操作:先去连接 47.93.4.197:6379,遇到IO则自动切换任务,去连接47.93.4.198:6379
        redis = await aioredis.create_redis_pool(address, password=password)
        # 网络IO操作:遇到IO会自动切换任务
        await redis.hmset_dict('car', key1=1, key2=2, key3=3)
        # 网络IO操作:遇到IO会自动切换任务
        result = await redis.hgetall('car', encoding='utf-8')
        print(result)
        redis.close()
        # 网络IO操作:遇到IO会自动切换任务
        await redis.wait_closed()
        print("结束", address)
    task_list = [
        execute('redis://47.93.4.197:6379', "root!2345"),
        execute('redis://47.93.4.198:6379', "root!2345")
    ]
    asyncio.run(asyncio.wait(task_list))

    5.2、异步MySQL

    pip install aiomysql

    示例1:

    import asyncio
    import aiomysql
    async def execute():
        # 网络IO操作:连接MySQL
        conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', db='mysql', )
        # 网络IO操作:创建CURSOR
        cur = await conn.cursor()
        # 网络IO操作:执行SQL
        await cur.execute("SELECT Host,User FROM user")
        # 网络IO操作:获取SQL结果
        result = await cur.fetchall()
        print(result)
        # 网络IO操作:关闭链接
        await cur.close()
        conn.close()
    asyncio.run(execute())

    示例2:

    import asyncio
    import aiomysql
    async def execute(host, password):
        print("开始", host)
        # 网络IO操作:先去连接 47.93.40.197,遇到IO则自动切换任务,去连接47.93.40.198:6379
        conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
        # 网络IO操作:遇到IO会自动切换任务
        cur = await conn.cursor()
        # 网络IO操作:遇到IO会自动切换任务
        await cur.execute("SELECT Host,User FROM user")
        # 网络IO操作:遇到IO会自动切换任务
        result = await cur.fetchall()
        print(result)
        # 网络IO操作:遇到IO会自动切换任务
        await cur.close()
        conn.close()
        print("结束", host)
    task_list = [
        execute('47.93.41.197', "root!2345"),
        execute('47.93.40.197', "root!2345")
    ]
    asyncio.run(asyncio.wait(task_list))

    5.3、FastAPI框架

    安装

    pip install fastapi
    pip install uvicorn (asgi内部基于uvloop)

    示例: luffy.py

    import asyncio
    import uvicorn
    import aioredis
    from aioredis import Redis
    from fastapi import FastAPI
    app = FastAPI()
    # 创建一个redis连接池
    REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", minsize=1, maxsize=10)
    @app.get("/")
    def index():
        """ 普通操作接口 """
        # 如果有两个用户并发访问此接口,用户A先执行结束并返回才能用户B执行,同步操作
        return {"message": "Hello World"}
    @app.get("/red")
    async def red():
        """ 异步操作接口 """
        
        print("请求来了")
        await asyncio.sleep(3)
        # 连接池获取一个连接
        conn = await REDIS_POOL.acquire()
        redis = Redis(conn)
        # 设置值
        await redis.hmset_dict('car', key1=1, key2=2, key3=3)
        # 读取值
        result = await redis.hgetall('car', encoding='utf-8')
        print(result)
        # 连接归还连接池
        REDIS_POOL.release(conn)
        return result
    if __name__ == '__main__':
        uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")

    六、总结

    最大的意义:通过一个线程利用其IO等待时间去做一些其他事情。

VPS购买请点击我

文章版权声明:除非注明,否则均为主机测评原创文章,转载或复制请以超链接形式并注明出处。

目录[+]