python 协程的简单使用

运行协程，asyncio 提供了三种主要机制

asyncio.run最高层级的入口点

对协程执行 await

使用asyncio.create_task()函数用来并发运行作为 asyncio任务的多个协程

多个顺序执行

importasyncioimporttime asyncio.runasyncdefsay_after(delay,what):awaitasyncio.sleep(delay)print(what)asyncdefmain():print(f"started at{time.strftime('%X')}")awaitsay_after(1,'hello')awaitsay_after(2,'world')print(f"finished at{time.strftime('%X')}")asyncio.run(main())

started at 20:15:44
hello
world
finished at 20:15:47

多个异步执行

importasyncioimporttimeasyncdefsay_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"started at{time.strftime('%X')}")# Wait until both tasks are completed (should take# around 2 seconds.)awaittask1awaittask2print(f"finished at{time.strftime('%X')}")asyncio.run(main())

started at 20:17:33
hello
world
finished at 20:17:35

通过上面的对比，可以看到第二个输出比第一个快了一秒

协程高级接口api

关于协程asyncio.gather(*aws,return_exceptions=False)

并发运行aws序列中的可等待对象
这个也是并行，上面的create_task可也并行。其实gather里面调用了create_task。这个loop是更低级的接口

如果return_exceptions为False(默认)，所引发的首个异常会立即传播给等待gather()的任务。aws序列中的其他可等待对象不会被取消并将继续运行。
如果return_exceptions为True，异常会和成功的结果一样处理，并聚合至结果列表。

这个方法会返回一个列表，用于存储可等待对象，也就是协程的返回结果。而且这个结果是跟任务的执行顺序一样

asyncdefrequest_get(session,url):asyncwithsession.get(url)asresponse:returnf'{url}内容:{awaitresponse.text()}'asyncdefmain():asyncwithaiohttp.ClientSession()assession:urls=["https://example.com/a","https://example.com/b","https://example.com/c"]tasks=[asyncio.wait_for(request_get(session,url),timeout=1)forurlinurls]results=awaitasyncio.gather(*tasks,return_exceptions=True)print(results)asyncio.run(main())

输出结果：

[ ‘https://example.com/a 内容: …’,‘https://example.com/b 内容: …’, ‘https://example.com/c 内容:…’]

当为return_exceptions为False，会直接报错，导致程序失败，所以需要增加异常处理。

当为return_exceptions为true，如果有异常，会存进返回的列表中。

asyncdefrequest_get(session,url):ifurl=='https://example.com/a':awaitasyncio.sleep(3)asyncwithsession.get(url)asresponse:returnf'{url}内容:{awaitresponse.text()}'asyncdefmain():asyncwithaiohttp.ClientSession()assession:urls=["https://example.com/a","https://example.com/b","https://example.com/c"]tasks=[asyncio.wait_for(request_get(session,url),timeout=1)forurlinurls]results=awaitasyncio.gather(*tasks,return_exceptions=True)print(results)foriinresults:print(i)asyncio.run(main())

输出结果：

[TimeoutError(),‘https://example.com/b 内容: …’, ‘https://example.com/c 内容:…’]

https://example.com/b 内容: …
https://example.com/c 内容: …

由上面的输出可以看到，TimeoutError()这个异常存进了结果列表。虽然通过for循环遍历了，但是程序并没有报错，而且没有输出，所以这就需要自己处理错误。for循环可以更改为一下所示：

foriinresults:ifisinstance(i,Exception):print(type(i))print(f"任务失败：{type(i).__name__}")else:print(i)

输出如下：

任务失败：TimeoutError
https://example.com/b 内容: …
https://example.com/c 内容: …

单个任务设置超时时间 asyncio.wait_for(aw,timeout)

aw是一个可等待对象，也就是协程
timeout是超时时间，可以为None，也可以为 float 或 int 型数值表示的等待秒数
如果发生超时，任务将取消并引发asyncio.TimeoutError
返回值是执行协程的返回值

多个任务设置超时时间 asyncio.wait(*aws,timeout，return_when=ALL_COMPLETED)

并发地运行
aws是一个可等待对象的可迭代对象。这跟上面的wait_for中的aw是不一样的。
timeout是超时时间，可以为None，也可以为 float 或 int 型数值表示的等待秒数
如果发生超时，任务将取消并引发asyncio.TimeoutError
返回两个 Task/Future 集合:(done, pending)（这个其实跟python的线程或者进程是类似的，可查看python 线程与多线程简单使用，他也有wait、as_completed方法。）

return_when的可选项如下：

注意：
1、这个wait与wait_for不一样，wait_for在超时后会抛错TimeoutError，但是wait方法不会。如下所示：

importaiohttpimportasyncioimporttimeasyncdefrequest_get(session,url):ifurl=='https://example.com/a':awaitasyncio.sleep(3)asyncwithsession.get(url)asresponse:returnf'{url}内容:{awaitresponse.text()}'asyncdefmain():asyncwithaiohttp.ClientSession()assession:urls=["https://example.com/a","https://example.com/b","https://example.com/c"]tasks=[request_get(session,url)forurlinurls]done,pending=awaitasyncio.wait(tasks,timeout=1)print('....................................')print(done)print('....................................')print(pending)

输出：

1766493515.1018946
…

{<Task finished name=‘Task-3’ coro=<request_get() done, defined at e:\code…> result=‘…’>,

<Task finished name=‘Task-4’ coro=<request_get() done, defined at e:\code…6> result=‘…’>}
…

{<Task pending name=‘Task-2’ coro=<request_get() running at e:\code\pycharmcode\test\asyhttp_test\01.py:8> wait_for=>}
1766493516.1151073

可以看出，时间从1766493515.1018946到1766493516.1151073，任务3、4都执行完成了，但是任务2没有完成。所有的输出没有任务error，都是task对象。这就引申出第二个注意事项
2、对未完成的任务做处理
上面的例子中，pending集合里还有任务，且它正在执行，如果我们不管，可能会造成资源消耗，内存泄漏等。
上面的例子简单，如果说有复杂的任务在规定时间内没有完成，然后不做处理，就可能会一直运行，我们也不知道。
我们可以运行task.cancel()方法来取消task对象，Task.cancel()不保证 Task 会被取消。示例如下：

fortaskinpending:task.cancel()try:awaittaskexceptasyncio.CancelledError:print("main(): cancel_me is cancelled now")

多个任务设置超时时间asyncio.``as_completed`(aws, ***,timeout=None)

并发地运行aws可迭代对象中的 可等待对象。返回一个协程的迭代器。 所返回的每个协程可被等待以从剩余的可等待对象的可迭代对象中获得最早的下一个结果。如果在所有 Future 对象完成前发生超时则将引发asyncio.TimeoutError

最早的下一个结果指的是结果与任务顺序不对照。例如三个任务1、2、3，执行后返回自己的数字，as_completed返回的可能是[1,3,2]或者[3,1,2]，谁先执行完，先返回谁。wait方法返回的结果只能是[1,2,3]

一点解惑

刚开始看协程，对于什么时间使用await不太理解。写出过下面的代码

asyncdefrequest_get(session,url):asyncwithsession.get(url)asresponse:returnf'{url}内容:{awaitresponse.text()}'asyncdefmain():asyncwithaiohttp.ClientSession()assession:urls=["https://example.com/a","https://example.com/b","https://example.com/c"]tasks=[awaitrequest_get(session,url)forurlinurls]done,pending=awaitasyncio.wait(tasks,timeout=1)

tasks = [await request_get(session, url) for url in urls]我在request_get前面使用了await，其实它的使用方法我在文章开头就写了，asyncio提供了三种主要机制来运行协程，其中就有await
所以按照上面我的错误写法，更改成下面的代码，执行你就会发现，没有执行wait，request_get也会执行。

asyncdefrequest_get(session,url):asyncwithsession.get(url)asresponse:returnf'{url}内容:{awaitresponse.text()}'asyncdefmain():asyncwithaiohttp.ClientSession()assession:urls=["https://example.com/a","https://example.com/b","https://example.com/c"]tasks=[awaitrequest_get(session,url)forurlinurls]# done, pending = await asyncio.wait(tasks, timeout=1) 将这一行删掉