Python运维之协程

目录

一、定义协程

二、并发

三、异步请求

协程是一种轻量级的线程，它通过保存和恢复寄存器上下文和栈来实现调度切换，从而保留函数执行的状态。

这种机制使得协程在处理I/O密集型任务时效率较高，因为它们可以在I/O操作期间让出CPU，以执行其他任务。与多线程相比，协程在同一线程内进行调用，减少了上下文切换的开销。

简而言之，协程通过在函数执行过程中灵活地让出和收回控制权，提高了程序的并发性能。

一、定义协程

python3.4加入协程的概念，以生成器对象为基础。Python3.5增加了async/await，下面以asyncio为基础介绍协程的使用。

 import asyncio
 import time
 ​
 async def task():
     print(f"{time.strftime('%H:%M:%S')} task 开始")
     time.sleep(2)
     print(f"{time.strftime('%H:%M:%S')} task 结束")
 ​
 coroutine = task()
 print(f"{time.strftime('%H:%M:%S')} 产生协程对象 {coroutine},函数并未被调用")
 loop = asyncio.get_event_loop()
 print(f"{time.strftime('%H:%M:%S')} 开始调用协程任务")
 start = time.time()
 loop.run_until_complete(coroutine)
 end = time.time()
 print(f"{time.strftime('%H:%M:%S')} 结束调用协程任务，耗时{end - start} 秒")

提示：首先引入asyncio，主要才可以使用async和await关键字（async定义一个协程，await用于临时挂起一个函数或方法的执行），接着使用async定义一个协程方法，然后直接调用该方法，但该方法没有被执行，而是返回一个coroutine协程对象。 使用get_event_loop()方法创建一个事件循环loop，并调用loop对象的run_until_complete()方法将协程注册到事件循环loop中，然后启动，这才完成执行。

我们还可以为任务绑定回调函数：

 import asyncio
 import time
 ​
 async def task():
     print(f"{time.strftime('%H:%M:%S')} task 开始")
     time.sleep(2)
     print(f"{time.strftime('%H:%M:%S')} task 结束")
     return "运行结束"
 ​
 def callback(task):
     print(f"{time.strftime('%H:%M:%S')} 回调函数开始执行")
     print(f"状态：{task.result()}")
 ​
 coroutine = task()
 print(f"{time.strftime('%H:%M:%S')} 产生协程对象 {coroutine},函数并未被调用")
 task = asyncio.ensure_future(coroutine)
 task.add_done_callback(callback)
 loop = asyncio.get_event_loop()
 print(f"{time.strftime('%H:%M:%S')} 开始调用协程任务")
 start = time.time()
 loop.run_until_complete(task)
 end = time.time()
 print(f"{time.strftime('%H:%M:%S')} 结束调用协程任务，耗时{end - start} 秒")

定义了一个协程方法和一个普通方法作为回调函数，回调函数接收一个参数是task对象，asyncio.ensure_future(coroutine)可以返回task对象，add_done_callback()为task对象增加一个回调任务。这样我们就定义好了一个coroutine对象和一个回调方法，执行的结果是当couroutine对象执行完毕之后，就去执行声明的callback方法。

二、并发

上述之定义了一个协程任务，如果要多次并尽可能提高效率，可以定义一个task列表，然后使用asyncio的wait()方法执行即可：

 import asyncio
 import time
 ​
 async def task():
     print(f"{time.strftime('%H:%M:%S')} task 开始")
     # 异步调用asynico.sleep(1):
     await asyncio.sleep(2)
     # time.sleep(2)
     time.sleep(2)
     print(f"{time.strftime('%H:%M:%S')} task 结束")
     return "运行结束"
 ​
 # 获取EventLoop:
 loop = asyncio.get_event_loop()
 # 执行coroutine
 tasks = [task() for _ in range(5)]
 start = time.time()
 loop.run_until_complete(asyncio.wait(tasks))
 loop.close()
 end = time.time()
 print(f"用时{end - start}")

关键字await后面的对象必须是以下类型之一：

• 一个原生coroutine对象
• 一个由types.coroutine()修饰的生成器，这个生成器可以返回coroutine对象

• 一个包含await方法的对象返回的一个迭代器

asyncio.sleep(2)是一个由coroutine修饰的生成器对象，表示等待2秒。

三、异步请求

以常用的网络请求为例，网络请求较多的就是I/O密集型任务。

启动一个简单的Web服务器

 from flask import Flask
 import time
 ​
 app =  Flask(__name__)
 ​
 @app.route('/')
 def index():
     time.sleep(3)
     return 'Hello world!'
 ​
 if __name__ == '__main__':
     app.run(threaded=True)      # 表明多线程模式启动

如果不开启多线程模式，那么同一时刻遇到多个请求时，只能顺次处理，这样即使我们使用协程异步请求这个服务，也只能一个一个排队。

 import asyncio
 import requests
 import time
 ​
 start = time.time()
 ​
 async def request():
     url = 'http://127.0.0.1:5000'
     print(f'{time.strftime("%H:%M:%S")} 请求 {url}')
     response = requests.get(url)
     print(f'{time.strftime("%H:%M:%S")} 得到响应 {response.text}')
 ​
 tasks = [asyncio.ensure_future(request()) for _ in range(5)]
 loop = asyncio.get_event_loop()
 loop.run_until_complete(asyncio.wait(tasks))
 ​
 end = time.time()
 print(f"耗时{end-start}")

耗时15秒，其实要实现异步处理，必须先有挂起的操作，当一个任务需要等待I/O结果时，可以挂起当前任务，让出CPU的控制权，转去执行其他任务，这样才能充分利用好资源。上述代码串行走，没有实现挂起

要实现异步，使用await将耗时等待的操作挂起让出控制权。当协程执行时遇到await，时间循环就会将本协程挂起，转去执行别的协程，直到其他的协程挂起或执行完毕，修改代码：

 import asyncio
 import requests
 import time
 ​
 async def get(url):
     return requests.get(url)
 ​
 async def request():
     url = 'http://127.0.0.1:5000'
     print(f'{time.strftime("%H:%M:%S")} 请求 {url}')
     response = await get(url)
     print(f'{time.strftime("%H:%M:%S")} 得到响应 {response.text}')
 ​
 start = time.time()
 tasks = [asyncio.ensure_future(request()) for _ in range(5)]
 loop = asyncio.get_event_loop()
 loop.run_until_complete(asyncio.wait(tasks))
 end = time.time()
 print(f"耗时{end-start}")

上述代码将请求页面的方法封装为一个coroutine读写，在request方法中尝试使用await挂起当前执行的I/O，发现还是15s，原来request不是异步请求，aiohttp是一个支持异步请求的库，将其配合使用即可实现异步请求操作：

 import asyncio
 import aiohttp
 import time
 ​
 now = lambda :time.strftime("%H:%M:%S")
 ​
 async def get(url):
     async with aiohttp.ClientSession() as session:  # 使用异步上下文管理器
         response = await session.get(url)
         result = await response.text()
         return result
     
 async def request():
     url = 'http://127.0.0.1:5000'
     print(f'{now()} 请求 {url}')
     result = await get(url)
     print(f'{now()} 得到响应 {result}')
 ​
 start = time.time()
 tasks = [asyncio.ensure_future(request()) for _ in range(5)]
 loop = asyncio.get_event_loop()
 loop.run_until_complete(asyncio.wait(tasks))
 end = time.time()
 print(f"耗时{end-start}")

运行时间只有3秒，扩大20倍还是3秒。可见，异步协程在爬虫项目值速度提升是非常可观了。