python的异步编程(基于asyncio)

见贤思齐

目录1.什么是异步编程2.Python有哪些可以实现异步编程3.异步编程(asyncio)3.1什么是事件循环3.2asyncio3.3async&await3.3.1 Task对象3.3.2 asyncio.Future对象3.3.3 futures.Future对象3.3.4异步迭代器3.2.5 异步上下文管理器3.2.6 uvloop4使用案例4.1异步Redis4.2异步MySQL1.什么是异步编程    异步编程允许程序在等待某些操作（如I/O操作、网络请求或定时器）完成时不阻塞（即不停止）主线程的执行，从而提高程序的效率和响应速度。在异步编程中，程序可以启动一个长时间运行的任务，然后继续执行其他任务，而无需等待该任务完成2.Python有哪些可以实现异步编程    1.greentlet:一个python的三方模块,通过将函数作为参数放到greenlet()中,并在函数内部通过switch()函数切换.        2.yield:想必大家都对这个不陌生, ython的生成器yield和yieldfrom也可以实现协程代码deffunc1():  yield1  yieldfromfunc2()  yield2deffunc2():  yield3  yield4f1=func1()foriteminf1:  print(item)    3.当然python还有其他的第三方模块可以实现异步编程，此处就不一一赘述3.异步编程(asyncio)        基于async&await关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术。3.1什么是事件循环        事件循环，可以把他当做是一个while循环，这个while循环在周期性的运行并执行一些任务，在特定条件下终止循环。其实也可以理解为就是创建一个死循环,比如while循环;不过这个死循环里面会周期性运行和执行任务,在特定条件下终止循环. 比如利用如下asyncio可以通过此来获取和创建事件循环.importasyncioloop=asyncio.get_event_loop()'运行运行3.2asyncio        在Python3.4之前官方未提供协程的类库，一般大家都是使用greenlet等其他来实现。在Python3.4发布后官方正式支持协程，即：asyncio模块。importasyncio@asyncio.coroutinedeffunc1():print(1)yieldfromasyncio.sleep(2)#遇到IO耗时操作，自动化切换到tasks中的其他任务print(2)@asyncio.coroutinedeffunc2():print(3)yieldfromasyncio.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))        注意：基于asyncio模块实现的协程比之前的要更厉害，因为他的内部还集成了遇到IO耗时操作自动切换的功能3.3async&await        async&await关键字在Python3.5版本中正式引入，基于它编写的协程代码其实就是上一示例的加强版，让代码可以更加简便.Python3.8之后@asyncio.coroutine装饰器就会被移除，推荐使用async&await关键字实现协程代码。importasyncioasyncdeffunc1():print(1)awaitasyncio.sleep(2)print(2)asyncdeffunc2():print(3)awaitasyncio.sleep(2)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模块和async&await关键字的方式，例如：在tonado、sanic、fastapi、django3中均已支持.        程序中，如果想要执行协程函数的内部代码，需要事件循环和协程对象配合才能实现，如:importasyncioasyncdeffunc():print("协程内部代码")#调用协程函数，返回一个协程对象。result=func()#方式一#loop=asyncio.get_event_loop()#创建一个事件循环#loop.run_until_complete(result)#将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。#方式二#本质上方式一是一样的，内部先创建事件循环然后执行run_until_complete，一个简便的写法。#asyncio.run函数在Python3.7中加入asyncio模块，asyncio.run(result)'运行运行        这个过程可以简单理解为：将协程当做任务添加到事件循环的任务列表，然后事件循环检测列表中的协程是否已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。        await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起当前协程（任务），当前协程（任务）挂起过程中事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。3.3.1 Task对象    Task用于并发调度协程,通过asyncio.create_task(协程对象)的方式创建Task对象,这样可以让协程加入事件循环中等待被调度执行,除了使用asyncio.create_task()函数以外，还可以用低层级的loop.create_task()或ensure_future()函数。不建议手动实例化Task对象。        本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。        注意：asyncio.create_task()函数在Python3.7中被加入。在Python3.7之前，可以改用低层级的asyncio.ensure_future()函数。importasyncioasyncdeffunc():print(1)awaitasyncio.sleep(2)print(2)return"返回值"asyncdefmain():print("main开始")#创建协程，将协程封装到Task对象中并添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。#在调用task_list=[asyncio.create_task(func(),name="n1"),asyncio.create_task(func(),name="n2")]print("main结束")#当执行某协程遇到IO操作时，会自动化切换执行其他任务。#此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done#如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。done,pending=awaitasyncio.wait(task_list,timeout=None)print(done,pending)asyncio.run(main())'运行运行        注意：asyncio.wait源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为[func(),func()]也是可以的。PS:importasyncioasyncdeffunc():print("执行协程函数内部代码")#遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。response=awaitasyncio.sleep(2)print("IO请求结束，结果为：",response)coroutine_list=[func(),func()]#错误：coroutine_list=[asyncio.create_task(func()),asyncio.create_task(func())]#此处不能直接asyncio.create_task，因为将Task立即加入到事件循环的任务列表，#但此时事件循环还未创建，所以会报错。#使用asyncio.wait将列表封装为一个协程，并调用asyncio.run实现执行两个协程#asyncio.wait内部会对列表中的每个协程执行ensure_future，封装为Task对象。done,pending=asyncio.run(asyncio.wait(coroutine_list))3.3.2 asyncio.Future对象        asyncio中的Future对象是一个相对更偏向底层的可对象，通常我们不会直接用到这个对象，而是直接使用Task对象来完成任务的并和状态的追踪.(Task是Futrue的子类)Future为我们提供了异步编程中的最终结果的处理(Task类也具备状态处理的功能)importasyncioasyncdefset_after(fut):awaitasyncio.sleep(2)fut.set_result("666")asyncdefmain():#获取当前事件循环loop=asyncio.get_running_loop()#创建一个任务（Future对象），没绑定任何行为，则这个任务永远不知道什么时候结束。fut=loop.create_future()#创建一个任务（Task对象），绑定了set_after函数，函数内部在2s之后，会给fut赋值。#即手动设置future任务的最终结果，那么fut就可以结束了。awaitloop.create_task(set_after(fut))#等待Future对象获取最终结果，否则一直等下去data=awaitfutprint(data)asyncio.run(main())'运行运行        扩展：支持await对象语法的对象课成为可等待对象，所以协程对象、Task对象、Future对象都可以被成为可等待对象。3.3.3 futures.Future对象        在Python的concurrent.futures模块中也有一个Future对象，这个对象是基于线程池和进程池实现异步操作时使用的对象。importtimefromconcurrent.futuresimportFuturefromconcurrent.futures.threadimportThreadPoolExecutorfromconcurrent.futures.processimportProcessPoolExecutordeffunc(value):time.sleep(1)print(value)pool=ThreadPoolExecutor(max_workers=5)#或pool=ProcessPoolExecutor(max_workers=5)foriinrange(10):fut=pool.submit(func,i)print(fut)'运行运行        两个Future对象是不同的,他们是为不同的应用场景而设计,例如:concurrent.futures.Future不支持await语法等.        在Python提供了一个将futures.Future对象包装成asyncio.Future对象的函数asynic.wrap_future。接下里你肯定问：为什么python会提供这种功能？        其实，一般在程序开发中我们要么统一使用asycio的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果协程的异步和进程池/线程池的异步混搭时，那么就会用到此功能了。importtimeimportasyncioimportconcurrent.futuresdeffunc1():#某个耗时操作time.sleep(2)return"SB"asyncdefmain():loop=asyncio.get_running_loop()#1.Runinthedefaultloop'sexecutor(默认ThreadPoolExecutor)#第一步：内部会先调用ThreadPoolExecutor的submit方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象#第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。#因为concurrent.futures.Future对象不支持await语法，所以需要包装为asycio.Future对象才能使用。fut=loop.run_in_executor(None,func1)result=awaitfutprint('defaultthreadpool',result)#2.Runinacustomthreadpool:#withconcurrent.futures.ThreadPoolExecutor()aspool:#result=awaitloop.run_in_executor(#pool,func1)#print('customthreadpool',result)#3.Runinacustomprocesspool:#withconcurrent.futures.ProcessPoolExecutor()aspool:#result=awaitloop.run_in_executor(#pool,func1)#print('customprocesspool',result)asyncio.run(main())'运行运行        应用场景：当项目以协程式的异步编程开发时，如果要使用一个第三方模块，而第三方模块不支持协程方式异步编程时，就需要用到这个功能，例如：importasyncioimportrequestsasyncdefdownload_image(url):#发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）print("开始下载:",url)loop=asyncio.get_event_loop()#requests模块默认不支持异步操作，所以就使用线程池来配合实现了。future=loop.run_in_executor(None,requests.get,url)response=awaitfutureprint('下载完成')#图片保存到本地文件file_name=url.rsplit('_')[-1]withopen(file_name,mode='wb')asfile_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)forurlinurl_list]loop=asyncio.get_event_loop()loop.run_until_complete(asyncio.wait(tasks))3.3.4异步迭代器    什么是异步迭代器                实现了__aiter__()和__anext__()方法的对象。__anext__必须返回一个awaitable对象。asyncfor会处理异步迭代器的__anext__()方法所返回的可等待对象，直到其引发一个StopAsyncIteration异常        什么是异步可迭代对象？                可在asyncfor语句中被使用的对象。必须通过它的__aiter__()方法返回一个asynchronousiterator。importasyncioclassReader(object):"""自定义异步迭代器（同时也是异步可迭代对象）"""def__init__(self):self.count=0asyncdefreadline(self):#awaitasyncio.sleep(1)self.count+=1ifself.count==100:returnNonereturnself.countdef__aiter__(self):returnselfasyncdef__anext__(self):val=awaitself.readline()ifval==None:raiseStopAsyncIterationreturnvalasyncdeffunc():#创建异步可迭代对象async_iter=Reader()#asyncfor必须要放在asyncdef函数内，否则语法错误。asyncforiteminasync_iter:print(item)asyncio.run(func())'运行运行        异步迭代器其实没什么太大的作用，只是支持了asyncfor语法而已。3.2.5 异步上下文管理器        此种对象通过定义__aenter__()和__aexit__()方法来对asyncwith语句中的环境进行控制importasyncioclassAsyncContextManager: def__init__(self):self.conn=connasyncdefdo_something(self):#异步操作数据库return666asyncdef__aenter__(self):#异步链接数据库self.conn=awaitasyncio.sleep(1)returnselfasyncdef__aexit__(self,exc_type,exc,tb):#异步关闭数据库链接 awaitasyncio.sleep(1)asyncdeffunc():asyncwithAsyncContextManager()asf:result=awaitf.do_something()print(result)asyncio.run(func())        这个异步的上下文管理器还是比较有用的，平时在开发过程中打开、处理、关闭操作时，就可以用这种方式来处理3.2.6 uvloop        ython标准库中提供了asyncio模块，用于支持基于协程的异步编程。uvloop是asyncio中的事件循环的替代方案，替换后可以使得asyncio性能提高。事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言。        在项目中想要使用uvloop替换asyncio的事件循环也非常简单，只要在代码中这么做就行。importasyncioimportuvloopasyncio.set_event_loop_policy(uvloop.EventLoopPolicy())#编写asyncio的代码，与之前写的代码一致。#内部的事件循环自动化会变为uvloopasyncio.run(...)        注意：知名的asgiuvicorn内部就是使用的uvloop的事件循环.但是uvloop严重依赖libuv,因此在windows系统上效果大打折扣4使用案例        为了更好理解，上述所有示例的IO情况都是以asyncio.sleep为例，而真实的项目开发中会用到很多IO的情况4.1异步Redis        当通过python去操作redis时，链接、设置值、获取值这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。importasyncioimportaioredisasyncdefexecute(address,password):print("开始执行",address)#网络IO操作：先去连接47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379redis=awaitaioredis.create_redis_pool(address,password=password)#网络IO操作：遇到IO会自动切换任务awaitredis.hmset_dict('car',key1=1,key2=2,key3=3)#网络IO操作：遇到IO会自动切换任务result=awaitredis.hgetall('car',encoding='utf-8')print(result)redis.close()#网络IO操作：遇到IO会自动切换任务awaitredis.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))4.2异步MySQL        当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。#!/usr/bin/envpython#-*-coding:utf-8-*-importasyncioimportaiomysqlasyncdefexecute(host,password):print("开始",host)#网络IO操作：先去连接47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379conn=awaitaiomysql.connect(host=host,port=3306,user='root',password=password,db='mysql')#网络IO操作：遇到IO会自动切换任务cur=awaitconn.cursor()#网络IO操作：遇到IO会自动切换任务awaitcur.execute("SELECTHost,UserFROMuser")#网络IO操作：遇到IO会自动切换任务result=awaitcur.fetchall()print(result)#网络IO操作：遇到IO会自动切换任务awaitcur.close()conn.close()print("结束",host)task_list=[execute('47.93.40.197',"root!2345"),execute('47.93.40.197',"root!2345")]asyncio.run(asyncio.wait(task_list))4.3FastAPI框架        FastAPI是一款用于构建API的高性能web框架，框架基于Python3.6+的typehints搭建。接下里的异步示例以FastAPI和uvicorn来讲解（uvicorn是一个支持异步的asgi）。#!/usr/bin/envpython#-*-coding:utf-8-*-importasyncioimportuvicornimportaioredisfromaioredisimportRedisfromfastapiimportFastAPIapp=FastAPI()REDIS_POOL=aioredis.ConnectionsPool('redis://47.193.14.198:6379',password="root123",minsize=1,maxsize=10)@app.get("/")defindex():"""普通操作接口"""return{"message":"HelloWorld"}@app.get("/red")asyncdefred():"""异步操作接口"""print("请求来了")awaitasyncio.sleep(3)#连接池获取一个连接conn=awaitREDIS_POOL.acquire()redis=Redis(conn)#设置值awaitredis.hmset_dict('car',key1=1,key2=2,key3=3)#读取值result=awaitredis.hgetall('car',encoding='utf-8')print(result)#连接归还连接池REDIS_POOL.release(conn)returnresultif__name__=='__main__':uvicorn.run("luffy:app",host="127.0.0.1",port=5000,log_level="info")