协程,也有人叫他微线程,是一种用户态内的上下文切换技术。简单说,就是一种能在单线程内实现多任务切换的技术,这种技术不是计算机提供的,是人为创造的。
比如:你一边蹲坑,一边玩手机,一边抽烟,同时进行多个任务!这就是协程的精髓。
我们来看一下几种实现协程的方法。
greenlet
库这是一种早期的协程方案,不过用得比较少了。
from greenlet import greenlet
def func1():
print(1) # 第二步
gr2.switch() # 切换到func2
print(2) # 第四步
def func2():
print(3) # 第三步
gr1.switch() # 切换到func1
print(4) # 第五步
gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch() # 第一步
yield
关键字yield
是我们很熟悉的用来搞生成器的关键字,没错,这玩意就可以搞简单的协程来玩,这种方法相对直观,也比较容易理解。
def func1():
yield 1
yield from func2()
yield 2
def func2():
yield 3
yield 4
f1 = func1()
# 别问为啥可以循环哟!
for item in f1:
print(item)
# 输出 1、3、4、2
asyncio
库在 Python/ target=_blank class=infotextkey>Python 3.4 及以后版本,可以使用 asyncio
库来实现协程。
import asyncio
async def func1():
print(1)
# 遇到三上老师的网络 IO 请求
awAIt asyncio.sleep(2) # 模拟IO耗时操作,开始切换到其他任务
print(2)
async def func2():
print(3)
# 又遇到苍老师的网络IO请求
await asyncio.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))
上述代码中要注意的地方:遇到 IO 阻塞,会自动切换。
async
和 await
关键字Python 3.5 及以后版本引入了 async
和 await
关键字,更加简洁方便。
import asyncio
async def func1():
print(1)
# 又找三上老师的网络 IO 请求
await asyncio.sleep(2)
print(2)
async def func2():
print(3)
# ...网络IO请求
await asyncio.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))
归纳总结上面的例子,他们的共同步骤都是:
定义协程函数:注意,协程函数的调用,获取到的是一个协程对象,而不是直接执行函数。
在协程函数中进行异步操作
创建事件循环(asyncio
):使用 asyncio
创建一个事件循环对象, asyncio.get_event_loop()
获取事件循环实例。
将协程函数添加到任务列表:将协程函数添加到任务列表中,等待事件循环调度执行。
运行事件循环
总结起来,这些代码段都是通过异步编程实现协程,使多个任务能够在非阻塞的情况下并发执行。无论是使用 greenlet
、yield
、asyncio
还是 async
/await
,它们都基于协程的概念,以不同的方式来实现异步任务的调度和执行
首先,让我们用普通的同步方式来下载图片。
import requests
def download_image(url):
print("开始下载:", url)
response = requests.get(url)
print("下载完成")
file_name = url.rsplit('_')[-1]
with open(file_name, mode='wb') as f:
f.write(response.content)
url_list = [
'https://api.dujin.org/bing/1920.php',
'https://api.dujin.org/bing/1920.php',
'https://api.dujin.org/bing/1920.php'
]
for item in url_list:
download_image(item)
接下来,我们使用异步编程来同时下载多张图片。
import aiohttp
import asyncio
async def fetch(session, url):
print("发送请求:", url)
async with session.get(url, verify_ssl=False) as response:
content = await response.content.read()
file_name = url.rsplit('_')[-1]
with open(file_name, mode='wb') as file_object:
file_object.write(content)
print('下载完成', url)
async def main():
async with aiohttp.ClientSession() as session:
url_list = [
'https://api.dujin.org/bing/1920.php',
'https://api.dujin.org/bing/1920.php',
'https://api.dujin.org/bing/1920.php'
]
tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
await asyncio.wait(tasks)
asyncio.run(main())
在接口自动化测试中,如果测试用例全部没有依赖关系,比如单接口测试中,所有参数独立定义好后,就可以考虑使用异步接口请求。这样相当于又小压了一波接口。
import aiohttp
import asyncio
async def test_api(endpoint):
async with aiohttp.ClientSession() as session:
async with session.get(endpoint) as response:
result = await response.json()
print(f"请求 {endpoint} 返回的结果:{result},断言.......")
async def main():
# 接口测试用例可以用各种方式读取出来,这里简单搞一搞
api_endpoints = [
"http://api.example.com/endpoint1?name=1&age=1",
"http://api.example.com/endpoint1?name=汉字&age=1",
"http://api.example.com/endpoint1?name=**&age=1"
...
]
tasks = [asyncio.create_task(test_api(endpoint)) for endpoint in api_endpoints]
await asyncio.wait
asyncio.run(main())
这样就可以简单的实现异步请求的接口自动化测试啦。