Python 异步编程async/await一、同步 vs 异步同步的问题同步代码按顺序执行遇到 I/O 阻塞时整个线程干等着。importtimedeftask_sync(name,seconds):print(f同步任务{name}开始)time.sleep(seconds)# 模拟 I/O 阻塞print(f同步任务{name}结束)# 顺序执行总耗时 2 3 5 秒task_sync(A,2)task_sync(B,3)异步的解决异步用await asyncio.sleep()代替time.sleep()让出控制权给其他协程。importasyncioasyncdeftask_async(name,seconds):print(f异步任务{name}开始)awaitasyncio.sleep(seconds)# 非阻塞等待print(f异步任务{name}结束)asyncdefmain():# 创建两个任务并发执行task_aasyncio.create_task(task_async(A,2))task_basyncio.create_task(task_async(B,3))awaittask_aawaittask_b asyncio.run(main())# 异步总耗时 ≈ 3 秒取最长而非 5 秒核心区别同步异步等待方式time.sleep()阻塞整个线程await asyncio.sleep()让出控制权多个任务顺序执行总耗时 和并发执行总耗时 ≈ 最长关键字无async def/awaitPython 异步编程async/await二、协程基础2.1 async def — 定义协程函数importasyncioasyncdefhello():print(hello)awaitasyncio.sleep(1)print(world)print(hello())# coroutine object hello at ... 不会立即执行调用协程函数返回协程对象必须放入事件循环才能执行。2.2 await — 等待可等待对象asyncdefgreet():print(开始打招呼)awaithello()# 等待 hello() 执行完print(打招呼结束)importasyncio asyncio.run(greet())规则await只能在async def函数内部使用。普通函数里不能写await。2.3 协程的返回值asyncdefadd(a,b):awaitasyncio.sleep(0.5)returnabasyncdefmain():resultawaitadd(3,5)print(相加结果:,result)# 8asyncio.run(main())Python 异步编程async/await三、事件循环与运行方式3.1 asyncio.run() — 推荐方式import asynciopythonpythonasync def main():await asyncio.sleep(1)print(“完成”)asyncio.run(main()) # 创建事件循环运行协程关闭循环### 3.2 手动管理事件循环旧方式 极少使用但了解即可 python import asyncio loop asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(main()) loop.close() python loop asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(main()) loop.close()3.3 create_task — 并发执行importasyncioasyncdefsay_after(delay,what):awaitasyncio.sleep(delay)print(what)asyncdefmain():task1asyncio.create_task(say_after(1,任务一))task2asyncio.create_task(say_after(1,任务二))awaittask1awaittask2# 两个任务并发执行总耗时 ≈ 1 秒asyncio.run(main())asyncio.create_task()将协程包装成 Task 对象立即调度执行不等待。Python 异步编程async/await四、并发执行工具4.1 asyncio.gather() — 并发等所有结果importasynciopythonasyncdeftask(name,delay):awaitasyncio.sleep(delay)returnf{name}的结果asyncdefdemo():resultsawaitasyncio.gather(task(A,2),task(B,1),task(C,3),)print(results)# [A的结果, B的结果, C的结果] 保持顺序asyncio.run(demo())适用场景需要等所有任务完成且关心返回值。4.2 asyncio.wait() — 按条件等待importasyncioasyncdeftask(name,delay):awaitasyncio.sleep(delay)print(f任务{name}完成)asyncdefmain():tasks[asyncio.create_task(task(X,2)),asyncio.create_task(task(Y,1)),]# 等任意一个完成就返回done,pendingawaitasyncio.wait(tasks,return_whenasyncio.FIRST_COMPLETED)print(f完成:{len(done)}个任务)# 取消未完成的fortinpending:t.cancel()asyncio.run(main())适用场景只需要第一个完成的结果其余可以取消。4.3 asyncio.as_completed() — 按完成顺序获取importasyncioasyncdeftask(name,delay):awaitasyncio.sleep(delay)returnf任务{name}tasks[asyncio.create_task(task(P,3)),asyncio.create_task(task(Q,1)),asyncio.create_task(task(R,2)),]forcoroinasyncio.as_completed(tasks):resultawaitcoroprint(完成一个:,result)# 输出顺序Q → R → P按完成快慢适用场景需要按完成顺序逐个处理结果。三种方式对比方式顺序等多久场景gather保持传入顺序等全部需要所有结果wait不保证等到条件满足只需部分结果as_completed按完成快慢逐个获取流式处理五、异步上下文管理器用async with管理异步资源如网络连接。5.1 自定义异步上下文管理器importasyncioclassAsyncResource:asyncdef__aenter__(self):print(异步进入资源)awaitasyncio.sleep(1)returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):print(异步退出资源)awaitasyncio.sleep(0.5)asyncdefdo_work(self):print(工作中...)asyncdefuse_resource():asyncwithAsyncResource()asres:awaitres.do_work()asyncio.run(use_resource())和同步对比同步异步__enter__/__exit____aenter__/__aexit__withasync with5.2 http实际应用中常用aiohttpimportaiohttpasyncdeffetch_url(url):asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url)asresponse:returnawaitresponse.text()asyncio.run(fetch_url(https://www.baidu.com))Python 异步编程async/await六、异步迭代器与异步生成器6.1 异步迭代器手写__aiter____anext__importasyncioclassAsyncCounter:def__init__(self,start,end):self.startstart self.endenddef__aiter__(self):returnselfasyncdef__anext__(self):ifself.startself.end:raiseStopAsyncIterationawaitasyncio.sleep(0.5)# 模拟异步操作valueself.start self.start1returnvalueasyncdefuse():asyncfornuminAsyncCounter(1,5):print(f异步迭代得到{num})asyncio.run(use())和同步对比同步异步__iter__/__next____aiter__/__anext__StopIterationStopAsyncIterationforasync for6.2 异步生成器用async defyield比手写迭代器简洁得多importasyncioasyncdefasync_range(start,end):foriinrange(start,end):awaitasyncio.sleep(0.3)yieldiasyncdefuse():asyncforvalinasync_range(10,15):print(异步生成器,val)asyncio.run(use())和同步生成器对比同步生成器异步生成器defasync defyieldyield不变forasync for七、异步队列生产者消费者importasyncio,randomasyncdefproducer(queue,id):foriinrange(3):itemf生产者{id}的产品{i}awaitqueue.put(item)print(f生产者{item})awaitasyncio.sleep(random.uniform(0.5,1.5))asyncdefconsumer(queue,id):whileTrue:itemawaitqueue.get()ifitemisNone:# 结束信号queue.task_done()breakprint(f消费者{id}消费{item})awaitasyncio.sleep(random.uniform(0.2,1.0))queue.task_done()asyncdefmain():queueasyncio.Queue(maxsize5)producers[asyncio.create_task(producer(queue,i))foriinrange(2)]consumers[asyncio.create_task(consumer(queue,i))foriinrange(3)]awaitasyncio.gather(*producers)# 等生产者完成for_inconsumers:awaitqueue.put(None)# 发送结束信号awaitasyncio.gather(*consumers)# 等消费者处理完asyncio.run(main())关键 API操作方法入队await queue.put(item)出队item await queue.get()标记完成queue.task_done()最大容量asyncio.Queue(maxsizeN)和同步 queue.Queue 对比同步异步queue.put(item)阻塞await queue.put(item)非阻塞queue.get()阻塞await queue.get()非阻塞八、异步 TCP 客户端底层 I/O不用第三方库用标准库asyncio.open_connection发 HTTP 请求asyncdeftcp_client(host,port,message):reader,writerawaitasyncio.open_connection(host,port)writer.write(message.encode())awaitwriter.drain()# 刷新缓冲区dataawaitreader.read(1024)writer.close()awaitwriter.wait_closed()returndata.decode()# 发 HTTP GET 请求requestGET /get HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\nresponseasyncio.run(tcp_client(httpbin.org,80,request))学习意义理解底层writer发数据/reader收数据机制P1 V3 的工具调用Function Calling本质上也是异步 HTTP 请求。Python 异步编程async/await九、异步 vs 多线程async/await多线程threading并发模型协程单线程多线程OS 调度切换成本极低用户态较高内核态适合场景I/O 密集型I/O 密集型GIL 限制不受影响受影响CPU 密集型慢调试难度较低较高竞态条件结论Python 里做 I/O 密集型并发优先用 async/await。