Python的协程异步IO（asyncio）详解

见贤思齐

一、协程简介1.1定义协程不是系统级线程，很多时候协程被称为“轻量级线程”、“微线程”、“纤程(fiber)”等。简单来说可以认为协程是线程里不同的函数，这些函数之间可以相互快速切换。协程和用户态线程非常接近，用户态线程之间的切换不需要陷入内核，但部分操作系统中用户态线程的切换需要内核态线程的辅助。协程是编程语言（或者lib）提供的特性（协程之间的切换方式与过程可以由编程人员确定），是用户态操作。协程适用于IO密集型的任务。常见提供原生协程支持的语言有：c++20、golang、python等，其他语言以库的形式提供协程功能，比如C++20之前腾讯的fiber和libco等等。1.2分类 协程有两种，一种无栈协程，python中以asyncio为代表，一种有栈协程，python中以gevent为代表，本文主要讲解asyncio线程。有栈线程无栈线程备注例子：lua threadpython geventC#yieldreturnC#async\awaitpythonasyncio无是否拥有单独的上下文：是否上下文包括寄存器、栈帧局部变量保存位置：栈堆无栈协程的局部变量保存在堆上，比如generator的数据成员。优点：1.每个协程有单独的上下文，可以在任意的嵌套函数中任何地方挂起此协程。2.不需要编译器做语法支持，通过汇编指令即可实现1.不需要为每个协程保存单独的上下文，内存占用低。2.切换成本低，性能更高。无缺点：1.需要提前分配一定大小的堆内存保存每个协程上下文，所以会出现内存浪费或者栈溢出。2.上下文拷贝和切换成本高，性能低于无栈协程。1.需要编译器提供语义支持，比如C#yieldreturn语法糖。2.只能在这个生成器内挂起此协程，无法在嵌套函数中挂起此协程。3.关键字有一定传染性，异步代码必须都有对应的关键字。作为对比，有栈协程只需要做对应的函数调用。无栈协程无法在嵌套函数中挂起此协程，有栈协程由于是通过保存和切换上下文包括寄存器和执行栈实现，可以在协程函数的嵌套函数内部yield这个协程并唤醒。二、python的asyncio协程详解2.1介绍asyncio是用来编写并发代码的库，使用async/await语法。asyncio被用作多个提供高性能Python异步框架的基础，包括网络和网站服务，数据库连接库，分布式任务队列等等。asyncio往往是构建IO密集型和高层级结构化网络代码的最佳选择。asyncio提供一组高层级API用于:并发地运行Python协程并对其执行过程实现完全控制;执行网络IO和IPC;控制子进程;通过队列实现分布式任务;同步并发代码;创建和管理事件循环，以提供异步API用于网络化,运行子进程，处理OS信号等等;使用transports实现高效率协议;通过async/await语法桥接基于回调的库和代码。2.2asyncio协程的使用（用的python3.8的语法）asyncio函数的源代码地址：https://github.com/python/cpython/tree/3.8/Lib/asyncio1）协程通过async/await语法进行声明，是编写asyncio应用的推荐方式。asyncio.run()函数用来运行最高层级的入口点"main()"函数。asyncio.sleep(delay,result=None,*,loop=None)函数用来阻塞指定的秒数。#coding=utf8importsysimportasyncioasyncdefmain():print('hello')awaitasyncio.sleep(1)print('world')asyncio.run(main())'运行运行2）事件循环函数（包括循环的创建、运行和停止）asyncio.get_running_loop()函数返回当前OS线程中正在运行的事件循环。asyncio.get_event_loop()函数获取当前事件循环。asyncio.set_event_loop(loop)函数将loop设置为当前OS线程的当前事件循环。asyncio.new_event_loop()函数创建一个新的事件循环。loop.run_until_complete(future)函数运行直到future(Future的实例)被完成。loop.run_forever()函数运行事件循环直到stop()被调用。loop.stop()函数停止事件循环。loop.is_running()函数返回True如果事件循环当前正在运行。loop.is_closed()函数如果事件循环已经被关闭，返回True。loop.close()函数关闭事件循环。loop.create_future()函数创建一个附加到事件循环中的asyncio.Future对象。loop.create_task(coro,*,name=None)函数安排一个协程的执行。返回一个Task对象。loop.set_task_factory(factory)函数设置一个task工厂，被用于loop.create_task()。loop.get_task_factory()函数返回一个任务工厂，或者如果是使用默认值则返回None。 示例1：#coding=utf8importsysimportasyncioasyncdeffun1():awaitasyncio.sleep(1)print('协程1')asyncdeffun2():awaitasyncio.sleep(1)print('协程2')loop=asyncio.get_event_loop()loop.run_until_complete(asyncio.wait([fun1(),fun2()]))loop.close()示例2：#coding=utf8importsysimportasyncioimporttime#一个对future进行赋值的函数asyncdefslow_operation(future,num):awaitasyncio.sleep(1)#给future赋值future.set_result('Future'+str(num)+'isdone!')defmain():loop=asyncio.get_event_loop()#创建一个futurefuture1=loop.create_future()#使用ensure_future创建Taskasyncio.ensure_future(slow_operation(future1,1))future2=loop.create_future()asyncio.ensure_future(slow_operation(future2,2))#gatherTasks，并通过run_uniti_complete来启动、终止looploop.run_until_complete(asyncio.gather(future1,future2))print(future1.result())print(future2.result())loop.close()if__name__=="__main__":main()'运行运行3）调度回调和延迟回调 loop.call_soon(callback,*args,context=None)函数安排callback在事件循环的下一次迭代时附带args参数被调用。回调按其注册顺序被调用。每个回调仅被调用一次。方法不是线程安全的。loop.call_soon_threadsafe(callback,*args,context=None)函数是call_soon()的线程安全变体。必须被用于安排来自其他线程的回调。loop.call_later(delay,callback,*args,context=None)函数安排callback在给定的delay秒（可以是int或者float）后被调用。loop.call_at(when,callback,*args,context=None)函数安排callback在给定的绝对时间戳的时间（一个int或者float）被调用，使用与loop.time()同样的时间参考。loop.time()函数根据时间循环内部的单调时钟，返回当前时间，float值。#coding=utf8importsysimportasynciofromthreadingimportThreadimporttimedefcallback(arg,loop):print('回调函数arg={}回调的时间time={}'.format(arg,loop.time()))asyncdeftask(loop):now=loop.time()print('时钟时间:{}'.format(time.time()))print('时事件循环时间:{}'.format(loop.time()))print('注册回调函数')loop.call_at(now+1,callback,'call_at1',loop)#等待1秒执行call_at函数loop.call_at(now+2,callback,'call_at2',loop)loop.call_later(3,callback,'call_later1',loop)#等待3秒执行call_later函数loop.call_later(4,callback,'call_later2',loop)loop.call_soon(callback,'call_soon',loop)#立即执行执行call_soon函数awaitasyncio.sleep(4)defmain():event_loop=asyncio.get_event_loop()try:print('进入事件循环监听')event_loop.run_until_complete(task(event_loop))#将事件循环对象传入task函数中finally:print('关闭事件循环监听')event_loop.close()if__name__=="__main__":main()'运行运行4）socket连接和Streams函数loop.create_connection(protocol_factory,host=None,port=None,*,ssl=None,family=0,proto=0,flags=0,sock=None,local_addr=None,server_hostname=None,ssl_handshake_timeout=None,happy_eyeballs_delay=None,interleave=None)函数打开一个流式传输连接，连接到由host和port指定的地址。loop.create_server(protocol_factory,host=None,port=None,*,family=socket.AF_UNSPEC,flags=socket.AI_PASSIVE,sock=None,backlog=100,ssl=None,reuse_address=None,reuse_port=None,ssl_handshake_timeout=None,start_serving=True)函数创建TCP服务(socket类型SOCK_STREAM)监听host地址的port端口。loop.create_unix_server(protocol_factory,path=None,*,sock=None,backlog=100,ssl=None,ssl_handshake_timeout=None,start_serving=True)函数与loop.create_server()类似但是专用于AF_UNIX套接字族。path是必要的Unix域套接字名称，除非提供了sock参数。抽象的Unix套接字,str,bytes和Path路径都是受支持的。loop.connect_accepted_socket(protocol_factory,sock,*,ssl=None,ssl_handshake_timeout=None)函数将已被接受的连接包装成一个传输/协议对。loop.sock_recv(sock,nbytes)函数从sock接收至多nbytes。socket.recv()的异步版本。loop.sock_recv_into(sock,buf)函数从sock接收数据放入buf缓冲区。模仿了阻塞型的socket.recv_into()方法。loop.sock_sendall(sock,data)函数将data发送到sock套接字。socket.sendall()的异步版本。loop.sock_accept(sock)函数接受一个连接。模仿了阻塞型的socket.accept()方法。loop.sock_sendfile(sock,file,offset=0,count=None,*,fallback=True)函数在可能的情况下使用高性能的os.sendfile发送文件。返回所发送的字节总数。asyncio.open_connection(host=None,port=None,*,loop=None,limit=None,ssl=None,family=0,proto=0,flags=0,sock=None,local_addr=None,server_hostname=None,ssl_handshake_timeout=None)函数建立网络连接并返回一对(reader,writer)对象。asyncio.start_server(client_connected_cb,host=None,port=None,*,loop=None,limit=None,family=socket.AF_UNSPEC,flags=socket.AI_PASSIVE,sock=None,backlog=100,ssl=None,reuse_address=None,reuse_port=None,ssl_handshake_timeout=None,start_serving=True)函数启动套接字服务。asyncio.open_unix_connection(path=None,*,loop=None,limit=None,ssl=None,sock=None,server_hostname=None,ssl_handshake_timeout=None)函数建立一个Unix套接字连接并返回(reader,writer)这对返回值。与open_connection()相似，但是操作在Unix套接字上。asyncio.start_unix_server(client_connected_cb,path=None,*,loop=None,limit=None,sock=None,backlog=100,ssl=None,ssl_handshake_timeout=None,start_serving=True)函数启动一个Unixsocket服务。与start_server()相似，但是是在Unix套接字上的操作。asyncio.StreamReader这个类表示一个提供api来从IO流中读取数据的读取器对象。reader.read(n=-1)函数读取n个byte.如果没有设置n,则自动置为-1，读至EOF并返回所有读取的byte。reader.readline()函数读取一行，其中“行”指的是以\n结尾的字节序列。如果读到EOF而没有找到\n，该方法返回部分读取的数据。如果读到EOF，且内部缓冲区为空，则返回一个空的bytes对象。reader.readexactly(n)函数精准读取n个bytes，不能超过也不能少于。reader.readuntil(separator=b'\n')函数从流中读取数据直至遇到分隔符成功后，数据和指定的separator将从内部缓冲区中删除(或者说被消费掉)。返回的数据将包括在末尾的指定separator。如果读取的数据量超过了配置的流限制，将引发LimitOverrunError异常，数据将留在内部缓冲区中并可以再次读取。如果在找到完整的separator之前到达EOF，则会引发IncompleteReadError异常，并重置内部缓冲区。IncompleteReadError.partial属性可能包含指定separator的一部分。reader.at_eof()函数如果缓冲区为空并且feed_eof()被调用，则返回True。asyncio.StreamWriter这个类表示一个写入器对象，该对象提供api以便于写数据至IO流中。writer.write(data)函数会尝试立即将data写入到下层的套接字。如果写入失败，数据会被排入内部写缓冲队列直到可以被发送。writer.writelines(data)函数会立即尝试将一个字节串列表（或任何可迭代对象）写入到下层的套接字。如果写入失败，数据会被排入内部写缓冲队列直到可以被发送。writer.close()函数会关闭流以及下层的套接字。writer.can_write_eof()函数如果下层的传输支持write_eof()方法则返回``True``，否则返回False。writer.write_eof()函数在已缓冲的写入数据被刷新后关闭流的写入端。writer.transport()函数返回下层的asyncio传输。writer.drain()函数等待直到可以适当地恢复写入到流。writer.is_closing()函数如果流已被关闭或正在被关闭则返回True。writer.wait_closed()函数等待直到流被关闭。server代码： #coding=utf8importasynciofromasyncioimportStreamReader,StreamWriterasyncdefecho(reader:StreamReader,writer:StreamWriter):data=awaitreader.read(1024)message=data.decode()addr=writer.get_extra_info('peername')print(f"Received{message}from{addr}")print(f"Send:{message}")writer.write(data)awaitwriter.drain()writer.close()asyncdefmain(host,port):server=awaitasyncio.start_server(echo,host,port)addr=server.sockets[0].getsockname()print(f'Servingon{addr}')asyncwithserver:awaitserver.serve_forever()asyncio.run(main("127.0.0.1",9999))client代码：#coding=utf8importasyncioasyncdeftcp_echo_client(message):reader,writer=awaitasyncio.open_connection('127.0.0.1',9999)print(f'Sendtoserver:{message}')writer.write(message.encode())awaitwriter.drain()data=awaitreader.read(1024)print(f'Receivedfromserver:{data.decode()}')writer.close()awaitwriter.wait_closed()if__name__=='__main__':whileTrue:send_msg=input("send:")asyncio.run(tcp_echo_client(send_msg))5）在线程或者进程池中执行代码loop.run_in_executor(executor,func,*args)函数安排在指定的执行器中调用func。#coding=utf8importasyncioimportconcurrent.futuresdefblocking_io():#Fileoperations(suchaslogging)canblockthe#eventloop:runtheminathreadpool.withopen('/dev/urandom','rb')asf:returnf.read(100)defcpu_bound():#CPU-boundoperationswillblocktheeventloop:#ingeneralitispreferabletorunthemina#processpool.returnsum(i*iforiinrange(5))asyncdefmain():loop=asyncio.get_running_loop()##Options:#1.Runinthedefaultloop'sexecutor:result=awaitloop.run_in_executor(None,blocking_io)print('defaultthreadpool',result)print("\n")#2.Runinacustomthreadpool:withconcurrent.futures.ThreadPoolExecutor()aspool:result=awaitloop.run_in_executor(pool,blocking_io)print('customthreadpool',result)print("\n")#3.Runinacustomprocesspool:withconcurrent.futures.ProcessPoolExecutor()aspool:result=awaitloop.run_in_executor(pool,cpu_bound)print('customprocesspool',result)asyncio.run(main())'运行运行6） asyncio.create_task(coro,*,name=None)函数用来将一个协程打包为一个Task排入日程准备执行，并返回Task对象。#coding=utf8importsysimportasyncioimporttimeasyncdefsay_after(delay,what):awaitasyncio.sleep(delay)print(what)asyncdefmain():task1=asyncio.create_task(say_after(1,'hello'))task2=asyncio.create_task(say_after(2,'world'))print(f"startedat{time.strftime('%X')}")#Waituntilbothtasksarecompleted(shouldtakearound2seconds.)awaittask1awaittask2print(f"finishedat{time.strftime('%X')}")asyncio.run(main())'运行运行7）错误处理APIloop.set_exception_handler(handler)函数将handler设置为新的事件循环异常处理器。loop.get_exception_handler()函数返回当前的异常处理器，如果没有设置异常处理器，则返回None。loop.default_exception_handler(context)函数默认的异常处理器。loop.call_exception_handler(context)函数调用当前事件循环的异常处理器。loop.get_debug()函数获取事件循环调试模式设置（bool）。loop.set_debug(enabled:bool)函数设置事件循环的调试模式。#coding=utf8importsysimportasynciodefhandle_exception(loop,context):print('Error:',context['message'])asyncdefmy_task():awaitasyncio.sleep(1)print('task1')loop=asyncio.get_event_loop()loop.set_exception_handler(handle_exception)loop.run_until_complete(my_task())loop.close()'运行运行8）Future asyncio.Future(*,loop=None)函数是一个Future代表一个异步运算的最终结果。线程不安全。asyncio.isfuture(obj)函数用来判断如果obj为一个asyncio.Future类的示例、 asyncio.Task类的实例或者一个具有 _asyncio_future_blocking 属性的对象，返回True。asyncio.ensure_future(obj,*,loop=None)函数创建新任务。asyncio.wrap_future(future,*,loop=None)函数将一个concurrent.futures.Future对象封装到asyncio.Future对象中。Future对象相关函数：fut.result()函数返回Future的结果。fut.set_result(result)函数将Future标记为完成并设置结果。fut.set_exception(exception) 函数将Future标记为完成并设置一个异常。fut.done()函数如果Future为已完成则返回True。fut.cancelled()函数是如果Future已取消则返回Truefut.add_done_callback(callback,*,context=None)函数添加一个在Future完成时运行的回调函数。fut.remove_done_callback(callback)函数从回调列表中移除callback。fut.cancel()函数取消Future并调度回调函数。fut.exception()函数返回Future已设置的异常。fut.get_loop()函数返回Future对象已绑定的事件循环。#coding=utf8importsysimportasyncioimporttime#定义一个协程asyncdefslow_operation(fut):awaitasyncio.sleep(1)fut.set_result(22)defdef_callback(fut):number=fut.result()print(number+1)defmain():#获得全局循环事件loop=asyncio.get_event_loop()#实例化期物对象fut=asyncio.Future()asyncio.ensure_future(slow_operation(fut))#执行回调函数fut.add_done_callback(def_callback)#loop的run_until_complete会将_run_until_complete_cb添加到future的完成回调列表中。而_run_until_complete_cb中会执行loop.stop()方法loop.run_until_complete(fut)#关闭事件循环对象loop.close()if__name__=="__main__":main()'运行运行9）asyncio.gather(*aws,loop=None,return_exceptions=False)函数用来并发运行aws序列中的可等待对象。如果aws中的某个可等待对象为协程，它将自动作为一个任务加入日程。#coding=utf8importsysimportasyncioasyncdeffactorial(name,number):f=1foriinrange(2,number+1):print(f"Task{name}:Computefactorial({i})...")awaitasyncio.sleep(1)f*=iprint(f"Task{name}:factorial({number})={f}")asyncdefmain():#Schedulethreecalls*concurrently*:awaitasyncio.gather(factorial("A",2),factorial("B",3),factorial("C",4),)asyncio.run(main())'运行运行10）asyncio.shield(aw,*,loop=None)函数用来保护一个可等待对象防止其被取消。#coding=utf8importsysimportasyncioasyncdeftask_func(number):awaitasyncio.sleep(1)print('函数执行成功：'+str(number))asyncdefcancel_task(task):awaitasyncio.sleep(0.2)was_cancelled=task.cancel()print(f'cancelled:{was_cancelled}')asyncdefmain():coro=task_func(1)task=asyncio.create_task(coro)shielded=asyncio.shield(task)asyncio.create_task(cancel_task(shielded))try:result=awaitshieldedprint(f'>got:{result}')exceptasyncio.CancelledError:print('shieldedwascancelled')awaitasyncio.sleep(1)print(f'shielded:{shielded}')print(f'task:{task}')asyncio.run(main())'运行运行11）asyncio.wait(aws,*,loop=None,timeout=None,return_when=ALL_COMPLETED)函数并发地运行aws可迭代对象中的可等待对象并进入阻塞状态直到满足return_when所指定的条件。return_when 指定此函数应在何时返回。它必须为以下常数之一：常数描述FIRST_COMPLETED函数将在任意可等待对象结束或取消时返回。FIRST_EXCEPTION函数将在任意可等待对象因引发异常而结束时返回。当没有引发任何异常时它就相当于 ALL_COMPLETED。ALL_COMPLETED函数将在所有可等待对象结束或取消时返回。#coding=utf8importsysimportasyncioasyncdefcoroutine_example(name):print('正在执行name:',name)awaitasyncio.sleep(1)print('执行完毕name:',name)loop=asyncio.get_event_loop()tasks=[coroutine_example('Zarten_'+str(i))foriinrange(3)]wait_coro=asyncio.wait(tasks)loop.run_until_complete(wait_coro)loop.close()12）asyncio.wait_for(aw,timeout,*,loop=None)函数等待aw可等待对象完成，指定 timeout 秒数后超时。#coding=utf8importsysimportasyncioasyncdefeternity():#Sleepforonehourawaitasyncio.sleep(3600)print('yay!')asyncdefmain():#Waitforatmost1secondtry:awaitasyncio.wait_for(eternity(),timeout=1)exceptasyncio.TimeoutError:print('timeout!')asyncio.run(main())'运行运行13） asyncio.run_coroutine_threadsafe(coro,loop)函数向指定事件循环提交一个协程。线程安全。#coding=utf8importsysimportasyncioimportthreadingasyncdefmain(i):whileTrue:awaitasyncio.sleep(1)print(i)asyncdefproduction_task():foriinrange(1,4):#将不同参数main这个协程循环注册到运行在线程中的循环，#thread_loop会获得一循环任务asyncio.run_coroutine_threadsafe(main(i),thread_loop)#注意：run_coroutine_threadsafe这个方法只能用在运行在线程中的循环事件使用defstart_loop(thread_loop):#运行事件循环，loop以参数的形式传递进来运行asyncio.set_event_loop(thread_loop)thread_loop.run_forever()if__name__=='__main__':#获取一个事件循环thread_loop=asyncio.new_event_loop()#将次事件循环运行在一个线程中，防止阻塞当前主线程，运行线程，同时协程事件循环也会运行threading.Thread(target=start_loop,args=(thread_loop,)).start()#将生产任务的协程注册到这个循环中loop=asyncio.get_event_loop()#运行次循环loop.run_until_complete(production_task())14） asyncio.current_task(loop=None)函数返回当前运行的Task实例，如果没有正在运行的任务则返回None。如果loop为None则会使用get_running_loop()获取当前事件循环。asyncio.all_tasks(loop=None)函数返回事件循环所运行的未完成的Task对象的集合。如果loop为None，则会使用get_running_loop()获取当前事件循环。#coding=utf8importsysimportasyncioimporttimeasyncdefmy_coroutine():task=asyncio.current_task()print(task)asyncdefmain():task1=asyncio.create_task(my_coroutine())task2=asyncio.create_task(my_coroutine())tasks=[task1,task2]awaitasyncio.gather(*tasks)asyncio.run(main())'运行运行15）asyncio.Task(coro,*,loop=None,name=None)函数一个与Future类似的对象，可运行Python协程。非线程安全。 Task对象相关函数：task.cancelled()函数如果Task对象被取消则返回True。task.done()函数如果Task对象已完成则返回True。task.result()函数返回Task的结果。task.exception()函数返回Task对象的异常。task.remove_done_callback(callback)函数从回调列表中移除callback。task.get_stack(*,limit=None)函数返回此Task对象的栈框架列表。task.print_stack(*,limit=None,file=None)函数打印此Task对象的栈或回溯。task.get_coro()函数返回由Task包装的协程对象。task.get_name()函数返回Task的名称。task.set_name(value)函数设置Task的名称。#coding=utf8importsysimportasyncioimporttimeasyncdefcancel_me():print('cancel_me():beforesleep')try:#Waitfor1hourawaitasyncio.sleep(3600)exceptasyncio.CancelledError:print('cancel_me():cancelsleep')raisefinally:print('cancel_me():aftersleep')asyncdefmain():#Createa"cancel_me"Tasktask=asyncio.create_task(cancel_me())#Waitfor1secondawaitasyncio.sleep(1)task.cancel()try:awaittaskexceptasyncio.CancelledError:print("main():cancel_meiscancellednow")asyncio.run(main())'运行运行16）基于生成器的协程（Python3.10之后已经弃用）基于生成器的协程是async/await语法的前身。它们是使用yieldfrom语句创建的Python生成器，可以等待Future和其他协程。@asyncio.coroutine用来标记基于生成器的协程的装饰器。asyncio.iscoroutine(obj)函数如果obj是一个协程对象则返回True。asyncio.iscoroutinefunction(func)函数如果func是一个协程函数则返回True。#coding=utf8importsysimportasyncioimporttime@asyncio.coroutine#标志协程的装饰器deftaskIO_1():print('开始运行IO任务1...')yieldfromasyncio.sleep(2)#假设该任务耗时2sprint('IO任务1已完成，耗时2s')returntaskIO_1.__name__@asyncio.coroutine#标志协程的装饰器deftaskIO_2():print('开始运行IO任务2...')yieldfromasyncio.sleep(3)#假设该任务耗时3sprint('IO任务2已完成，耗时3s')returntaskIO_2.__name__asyncio.run(taskIO_1())asyncio.run(taskIO_2())