python——异步编程原理

摘要

主要介绍python中相关的异步编程的原理

协程函数

在定义的时候使用async def 函数名():

协程对象 执行协程函数（）得到的协程对象

#协程函数
async def func():
    pass

#result 协程对象 内代码是不会执行的
result=func()

注意︰执行协程函数创建协程对象，函数内部代码不会执行。如果想要运行协程函数内部代码，必须要讲协程对象交给事件循环来处理。

import asyncio

#协程函数
async def func():
    pass

#result 协程对象 内代码是不会执行的
result=func()


loop = asyncio.get_event_loopO
loop.run_until_complete(result)#这样才能运行起来协程函数


在python 3.7的时候
asyncio.run(result)

事件循环

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

#伪代码

任务列表=〔妊务1，任务2，任务3,...]

while True:
    可执行的任务列表，已完成的任务列表=去任务列表中检查所有的任务，将'可执行'和'已完成'的任务返回

    for就绪任务in已准备就绪的任务列表:

        执行已就绪的任务

    for已完成的任务 in已完成的任务列表:

        在任务列表中移除已完成的任务

    如果任务列表中的任务都已完成，则终止循环

import asyncio

#去生成或获取一个事件循环
loop = asyncio.get_event_loopO

#将任务放到`任务列表`
loop.run_until_complete(任务)

await关键字

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

import asyncio

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

asyncio.run( funco )

#python3.7的环境下在才能允许这个操作

import asyncio

async def others():
    print("start")
    await asyncio.sleep(2)
    print( 'end ' )
    return '返回值'

async def func():
    print("执行协程函数内部代码")
    #遇到To操作挂起当前协程（任务)，等Io操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程(任务)。
    response =await others()
    print("Io请求结束，结果为: "，response)

asyncio.run( func() )

#python3.7的环境下在才能允许这个操作

import asyncio

async def others():
    print("start")
    await asyncio.sleep(2)
    print( 'end ' )
    return '返回值'

async def func():
    print("执行协程函数内部代码")
    #遇到To操作挂起当前协程（任务)，等Io操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程(任务)。
    response1 =await others()
    print("Io请求结束，结果为: "，response1)

    response2 =await others()
    print("Io请求结束，结果为: "，response2)

asyncio.run( func() )

await就是等待对象的值得到结果之后再继续向下走。

Task对象

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()函数。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: Task对象.py
@time: 2021/9/12 17:50
"""
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())

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: Task对象.py
@time: 2021/9/12 17:50
"""
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())
    task_list = [asyncio.create_task(func()), asyncio.create_task(func())]
    print("main结束")
    # 当执行某协程遇到Io操作时，会自动化切换执行其他任务。# 此处的await是等待相对应的协程全都执行完毕并获取结果
    # ret1 = await task1
    # ret2 = await task2
    done, pedging = await asyncio.wait(task_list, timeout=None)
    print(done, pedging)

asyncio.run(main())

这样的代码会出错

import asyncio

async def func():
    print(1)
    await asyncio.s1eep(2)
    print(2)
    return "返回值"

task_list = [asyncio.create_task(func(), name='n1 '), asyncio.create_task(func(), name='n2 ')]

done, pending = asyncio.run(asyncio.wait(task_list))
print(done)

import asyncio

async def func():
    print(1)
    await asyncio.s1eep(2)
    print(2)
    return "返回值"

task_list = [func(), func()]

done, pending = asyncio.run(asyncio.wait(task_list))
print(done)

asynico.Future对象

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

async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("6666")

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())

concurrent.futures.Future对象

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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: concurrent_Future.py
@time: 2021/9/12 18:41
"""

import time
from concurrent.futures import Future
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor

def func(value):
    time.sleep()
    print(value)

# 创建线程池
pool = ThreadPoolExecutor(max(5))

# 创建进程池
pool2 = ProcessPoolExecutor(max(5))

for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

if __name__ == '__main__':
    pass

以后写代码可能会存在交叉时间。例如: crm项目80%都是基于协程异步编程+MySQL(不支持)【线程、进程做异步编程】

 实际的案例

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: 异步和非异步的支持.py
@time: 2021/9/12 18:58
"""

import asyncio
import requests

async def download_images(url):
    # 发送网络请求，下载图片(遇到网络下载图片的Io请求，自动化切换到其他任务)
    print("开始下载", url)
    loop = asyncio.get_event_loop()
    # request 模板默认是不支持的异步操作的，所以在使用线程池来配合实现
    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)
    print("图片保存完成")

if __name__ == "__main__":
    url_list = [
        'https://img.alicdn.com/imgextra/i3/127167800/O1CN01lywipS27US0reYkHE_!!0-saturn_solar.jpg_468x468q75.jpg',
        'https://img.alicdn.com/imgextra/i2/127167800/O1CN01M4cuWt27US1BH6Ke6_!!0-saturn_solar.jpg_468x468q76.jpg',
        'https://img.alicdn.com/imgextra/i4/206720011/O1CN01HKfJ251Bx53r6kuqZ_!!0-saturn_solar.jpg_468x468q78.jpg'
    ]
    tasks = [download_images(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

异步迭代器

什么是异步迭代器：实现了(__aiter__()和__anext__() 方法的对象。___anext__必须返回一个awaitable对象。async_for 会处理异步迭代器的__anext__()方法所返回的可等待对象，直到其引发一个StopAsyncIteration 异常。什么是异步可迭代对象:可在 async for 语句中被使用的对象。必须通过它的__aiter__()方法返回一个asynchronous iterator。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: 异步迭代器.py
@time: 2021/9/12 19:19
"""
import asyncio

"""自定义的迭代器 同时也是异步可迭代对象"""

class Reader(object):

    def __int__(self):
        self.count = 0

    async def readline(self):
        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的函数的内部实现
    async for item in obj:
        print(item)

asyncio.run(func())

if __name__ == '__main__':
    pass

异步上下文管理器

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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: 异步管理器.py
@time: 2021/9/12 20:13
"""

import asyncio

class AsyncConttextManager:
    def __int__(self, conn):
        self.conn = conn

    async def do_somethiing(self):
        # 异步操作数据库
        return 666

    async def __aenter__(self):
        # 异步连接数据库
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 异步的关闭的数据库连接
        await asyncio.sleep(1)

async def func():
    async with AsyncConttextManager() as f:
        result = await f.do_somethiing()
        print(result)

asyncio.run(func())

if __name__ == '__main__':
    pass

uvloop

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

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