Agent Day3-4 Python asyncio学习

async 异步编程

1. 协程

  • 协程不是计算机提供，程序员人为创造
  • 协程，也可以叫做微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块相互切换执行。
def fun1():
    print(1)
    ...
    print(2)

def fun2():
    print(3)
    ...
    print(4)

fun1()
fun2()

实现协程的几种方法：

  • greenlet,早期模块。
  • yield关键字。
  • asyncio装饰器（py3.4）
  • async、await关键字（py3.5）[推荐]
1.1 greenlet实现协程
from greenlet import greenlet;
def func1():
    print(1)
    gr2.switch()
    print(2)
    gr2.switch()

def func2():
    print(3)
    gr1.switch()
    print(4)
gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch() # 第一步执行 func1 函数
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)
1.3 asyncio
# 在Python3.4及之后的版本
import asyncio
@asyncio.coroutine #表示下面的方法是协程方法
def fun1():
    print("1")
    yield from asyncio.sleep(2)
    print("2")

@asyncio.coroutine
def fun2():
    print("3")
    yield from asyncio.sleep(2)
    print("4")

 tasks = [
     asyncio.ensure_future(fun1()),
     asyncio.ensure_future(fun2())
 ]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
# 注意：遇到IO阻塞自动切换
1.4 async&await关键字

在python3.5及之后的版本

import asyncio
async def fun1():
    print("1")
    await asyncio.sleep(2) # 遇到IO耗时操作时，自动化切换到tasks中的其他任务
    print("2")


async def fun2():
    print("3")
    await asyncio.sleep(2)
    print("4")

tasks = [
     asyncio.ensure_future(fun1()),
     asyncio.ensure_future(fun2())
 ]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
# 注意：遇到IO阻塞自动切换

2. 协程的意义

意义：在一个线程中，如果遇到IO等待，线程不会等待，会执行其他代码

  • 同步和异步的区别
    同步是指完成一个任务之后再完成另一个任务
    异步是指再进行的一个任务的时候不等结果，又开始了下一个任务。

3. 异步编程

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

伪代码
任务列表 = [ 任务1，任务2，任务3，… ]
while True:
可执行的任务列表，已完成的任务列表 = 去任务列表中检查所有的任务，将’可执行’和’已完成’的任务返回
for 就绪任务 in 可执行的任务列表:
执行已就绪的任务
for 已完成的任务 in 已完成的任务列表:
在任务列表中移除 已完成的任务
如果 任务列表 中的任务都已完成，则终止循环

import asyncio
# 去生成或获取一个事件循环（event loop）
loop = asyncio.get_event_loop()
# 将tasks放入循环
loop.run_until_complete(asyncio.wait(tasks))
3.2快速上手
  • 协程函数:定义函数时候async def函数名
  • 协程对象:执行协程函数（）得到的协程对象 。
import asyncio
async def func():
    pass
result = func()
  • 注意：执行协程函数创建协程对象，函数内部代码不会执行。
  • 如果想要运行协程函数内部代码，必须要将协程对象交给事件循环来处理
import async
async def func():
    print("快来呀")

result = func()

# loop= asyncio.get_event_loop()
# loop.run_until_complete
asyncio.run(result) #python3.7之后
3.3 await

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

  • 示例一
async def x():
    print("come")
    response=await asyncio.sleep(2) #到这一步会先执行Event loop里面的事件，如果Event loop里面没有事件，就等待2秒
    print("结束",response)

asyncio.run(x())
  • 示例二
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() )
  • 示例三
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() )
#await会等待对象的值得到结果之后再往下走
3.4 Task对象
  • Tasks用于并发调度协程,用asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程对象加入到事件循环中等待被调度执行。简单来说，就是在Event loop中加入多个任务。
  • 示例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() )
# 结果是1. `print("main开始")`
# 2. 创建 `task1`、`task2` 两个任务，交给事件循环
# 3. `print("main结束")`
# 4. 遇到 `await task1` → 进入 `func`，打印 `1`，碰到 `await asyncio.sleep(2)`
#    - 此时**协程挂起，事件循环切换去跑 task2**，task2 也打印 `1`，同样进入 sleep (2)
#    - 两个 sleep **并发计时**，一起等 2 秒
# 5. 2 秒时间到，两个任务继续往下，先后打印 `2`
# 6. 拿到返回值，打印结果
  • 示例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)
    #done是已经跑完的任务，pending是还没跑完的任务

asyncio.run( main() )
  • 练习
import asyncio
async def work(x):
    await asyncio.sleep(1)
    print(x)

async def main():
    work(1)
    t = asyncio.create_task(work(2))
    await t

asyncio.run(main())
# 耗时1s，打印2
3.5 asyncio.Future对象

  • Future 是一种特殊的 低层级 可等待对象，表示一个异步操作的 最终结果。

  • 当一个 Future 对象 被等待，这意味着协程将保持等待直到该 Future 对象在其他地方操作完毕。

  • 在asyncio中需要 Future 对象以便允许通过 async/await 使用基于回调的代码。

  • 通常情况下没有必要在应用层级的代码中创建 Future 对象。

  • Future 对象有时会由库和某些 asyncio API 暴露给用户，用作可等待对象。

  • Task继承Future，Task对象内部await结果的处理是基于Future对象来的

  • 示例一

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务（Future对象），这个任务什么都不干。
    fut = loop.create_future()

    # 等待任务最终结果（Future对象），没有结果则会一直等下去。
    await fut

asyncio.run( main() )
  • 示例二
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.6 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)
  • 以后写代码时可能会存在交叉时间。例如：crm项目大多都是基于协程异步编程+MySQL(不支持协程)【用线程、进程做异步编程】
3.7 asyncio.gather()
  • gather(*aws, return_exceptions=False)
    1. *aws：位置参数，放一堆可等待对象（协程、Task、Future）
    1. return_exceptions=False（默认）
    • 默认 False：只要任意一个任务抛出异常，gather 整体直接抛出异常，剩下没跑完的任务会继续执行，但是不会收集结果
    • 改成 True：把异常当成返回值放进结果列表里，不会中断整个 gather，其他任务照常跑完
    1. 作用：并发批量执行多个协程，统一收集所有返回结果，等全部任务完成后，把结果按传入顺序打包成列表返回
  • 示例一
import asyncio
async def query_llm(name):
    await asyncio.sleep(1)
    print(name)

async def main():
    await asyncio.gather(
        query_llm("RAG"),
        query_llm("Agent"),
        query_llm("Workflow")
    )

asyncio.run(main()) #总耗时1s
  • 示例二
import asyncio

async def foo(num):
    await asyncio.sleep(num)
    return f"结果:{num}"

async def main():
    # 同时启动3个协程，并发跑
    res_list = await asyncio.gather(
        foo(1),
        foo(2),
        foo(1)
    )
    print(res_list)

asyncio.run(main())
# 运行总耗时 ≈ **2 秒**（最长的那个任务耗时），而不是 1+2+1=4s。
# 返回的列表顺序**和你传给 gather 的顺序完全一致**，和谁先跑完没关系！
# 输出：`['结果:1', '结果:2', '结果:1']`

并发、串行与并行的区别

    1. 并发
      并发：一个 CPU 核心，快速切换任务，看起来同时跑；
    1. 串行
      串行：顺序执行，任务排队，一个任务做完才能做下一个。
    1. 并行
      并行：多个 CPU 核心，任务真的在同一时刻一起跑。
    1. 通俗比喻
  • 并发：厨师先炒两下菜 A，关火，去切菜 B；切一会儿，回头再炒菜 A。来回切换。同一时间他只在做一件事，只是切换快，感觉两道菜同时在做。
  • 并行：有2 个厨师（2 核 CPU），厨师 1 炒菜 A，厨师 2 炒菜 B，两个人同一时刻一起动手。
  • 串行：先把第一个菜炒完，才能炒第二个菜。顺序执行，任务排队

总结：花了挺长时间，看了几个视频，发现有好些东西不清楚。我觉得后面还是看看官方文档快得多，或者两个配合，实在不明白的话，问问ai，到实际做项目时遇到问题再解决。

# python # 学习
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

Codex-CLI-安装与使用详解

在开始安装之前，我们先花点时间搞清楚我们即将安装的这个工具是什么。很多人在搜索引擎里搜 “Codex”，可能会得到完全不同的结果——这是因为 “Codex” 这个名字背后有好几个不同的东西。环境检查→ Node、Git、npm 三件套；安装→ npm / Homebrew / 源码，三选一；登录→ OpenAI 账号授权，一套状态复用；四大形态→ CLI、App、云端、IDE 插件，按场景选；高频

avatar 2048 AI社区
cover

从 GLM 到 GPT，HarnessPro 如何用 RSI 降低 Agent 成本

avatar 2048 AI社区
cover

【元宝搜索开发者征文大赛】如何免费实现新闻热点的自动获取？用“元宝搜索”写一个可以自动获取全网热点Agent

avatar 2048 AI社区