asyncio中低阶API的种类很多,涉及到开发的5个方面,这篇文章主要为大家详细介绍了这些低阶API的具体使用,感兴趣的小伙伴可以学习一下

1.低阶API介绍
asyncio中低阶API的种类很多，涉及到开发的5个方面。包括：

获取事件循环
事件循环方法集
传输
协议
事件循环策略
本篇中只讲解asyncio常见常用的函数，很多底层函数如网络、IPC、套接字、信号等不在本篇范围。

2.获取事件循环
事件循环是异步中重要的概念之一，用于驱动任务的执行。包含的低阶API如下：

get_event_loop()创建一个事件循环，用于驱动协程的执行

import asyncio
 
async def demo(i):
    print(f"hello {i}")
 
def main():
    loop = asyncio.get_event_loop()
    print(loop._selector)
    task = loop.create_task(demo(1))
    loop.run_until_complete(task)
 
main()

结果：

<selectors.KqueueSelector object at 0x104eabe20>
hello 1

可以通过loop._selector属性获取到当前事件循环使用的是kqueue模型

获取循环

import asyncio
 
async def demo(i):
    res = asyncio.get_running_loop()
    print(res)
    print(f"hello {i}")
 
 
def main():
    loop = asyncio.get_event_loop()
    task = loop.create_task(demo(1))
    loop.run_until_complete(task)
main()

结果：

<_UnixSelectorEventLoop running=True closed=False debug=False>
hello 1

推荐使用asyncio.run 创建事件循环，底层API主要用于库的编写。

3.生命周期
生命周期是用于管理任务的启停的函数，如下：

运行一个期程/任务/可等待对象直到完成。run_until_complete的参数是一个futrue对象。当传入一个协程，其内部会自动封装成task。run_until_complete()是会自动关闭事件循环的函数，区别于run_forever()是需要手动关闭事件循环的函数。

import asyncio 
 
 
async def demo(i):
    print(f"hello {i}")
 
 
def main():
    loop = asyncio.get_event_loop()
     
    task = loop.create_task(demo(1))
 
    # 传入的是一个任务
    loop.run_until_complete(task)
 
    # 传入的是一个协程也可以
    loop.run_until_complete(demo(20))
 
 
main()

结果：

hello 1
hello 20

4.调试

5.调度回调函数
在异步编程中回调函数是一种很常见的方法，想要在事件循环中增加一些回调函数，可以有如下方法：

call_soon函数原型：

loop.call_soon(callback, *args, context=None)
示例：

import asyncio
 
async def my_coroutine():
    print("协程被执行")
 
async def other_coro():
    print("非call_soon调用")
 
def callback_function():
    print("回调函数被执行")
 
 
# 创建一个事件循环
loop = asyncio.get_event_loop()
 
# 使用create_task包装协程函数，并调度执行
loop.call_soon(loop.create_task, my_coroutine())
 
# 调度一个常规函数以尽快执行
loop.call_soon(callback_function)
 
# 启动一个事件循环
task = loop.create_task(other_coro())
loop.run_until_complete(task)

结果：

回调函数被执行
非call_soon调用
协程被执行

结果分析：

call_soon调用普通函数直接传入函数名作为参数，调用协程函数需要讲协程通过loop.create_task封装成task。

6.线程/进程池

run_in_executor

import asyncio
import concurrent.futures
 
def blocking_function():
    # 模拟一个阻塞的操作
    import time
    time.sleep(2)
    return "阻塞函数返回"
 
async def async_function2():
    print("async_function2 start")
    await asyncio.sleep(1)
    print("async_function2 end")
 
async def async_function():
    print("异步函数开始执行。。。")
 
    print("调用同步阻塞函数")
    # 使用run_in_executor调度执行阻塞函数
    result = await loop.run_in_executor(None, blocking_function)
 
    print(f"获取同步函数的结果: {result}")
 
# 创建一个事件循环
loop = asyncio.get_event_loop()
 
# 运行异步函数
loop.run_until_complete(asyncio.gather(async_function(), async_function2()))

结果：

异步函数开始执行。。。
调用同步阻塞函数
async_function2 start
async_function2 end
获取同步函数的结果: 阻塞函数返回

结果分析：

通过事件循环执行任务async_function，在async_function中通过loop.run_in_executor调用同步阻塞函数blocking_function，该阻塞函数没有影响事件循环中另一个任务async_function2的执行。

await loop.run_in_executor(None, blocking_function)中None代表使用的是默认线程池，也可以替换成其他线程池。

使用自定义线程池和进程池

import asyncio
import concurrent.futures
 
def blocking_function():
    # 模拟一个阻塞的操作
    import time
    time.sleep(2)
    return "阻塞函数返回"
 
async def async_function():
    print("异步函数开始执行。。。")
 
    print("调用同步阻塞函数")
 
    # 线程池
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, blocking_function)
        print('线程池调用返回结果：', result)
 
    # 进程池
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, blocking_function)
        print('进程池调用返回结果：', result)
 
if __name__ == '__main__':
    # 创建一个事件循环
    loop = asyncio.get_event_loop()
 
    # 运行异步函数
    loop.run_until_complete(async_function())

结果：

异步函数开始执行。。。
调用同步阻塞函数
线程池调用返回结果： 阻塞函数返回
进程池调用返回结果： 阻塞函数返回

结果分析：

通过线程池concurrent.futures.ThreadPoolExecutor()和进程池concurrent.futures.ProcessPoolExecutor()执行阻塞函数。

7.任务与期程

import asyncio
 
 
def foo(future, result):
    print(f"此时future的状态:{future}")
    future.set_result(result)
    print(f"此时future的状态:{future}")
 
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
 
    # 手动创建future对象
    all_done = loop.create_future()
 
    # 设置一个回调函数用于修改设置future的结果
    loop.call_soon(foo, all_done, "Future is done!")
 
    result = loop.run_until_complete(all_done)
 
    print("返回结果", result)
    print("获取future的结果", all_done.result())

结果：

此时future的状态:<Future pending cb=[_run_until_complete_cb() at /Users/lib/python3.10/asyncio/base_events.py:184]>
此时future的状态:<Future finished result='Future is done!'>
返回结果 Future is done!
获取future的结果 Future is done!

结果分析：

future设置结果之后之后，future对象的状态就从pending变成finished状态。如果一个future没有手动设置结果，那么事件循环就不会停止。

create_task将协程封装成一个task对象，事件循环主要操作的是task对象。协程没有状态，而task是有状态的。

import asyncio 
 
 
async def demo(i):
    print(f"hello {i}")
    await asyncio.sleep(1)
 
def main():
    loop = asyncio.get_event_loop()
 
    # 将携程封装成task，给事件使用
    task = loop.create_task(demo(1))
 
    loop.run_until_complete(task)
 
main()
>>> 
hello 1

asyncio.create_task 和 loop.create_task的区别：

两者实现的功能都是一样的，将协程封装成一个task，让协程拥有了生命周期。区别仅仅在于使用的方法。asyncio.create_task 是高阶API，不需要创建事件循环，而loop.create_task需要先创建事件循环再使用该方法。

到此这篇关于python异步编程之asyncio低阶API的使用详解的文章就介绍到这了,更多相关python asyncio低阶API内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！

原文链接：https://www.cnblogs.com/goldsunshine/p/17950911