python协程测试限流

不言

pythontest

发布于：2020年6月17日

次浏览

背景

项目组使用sentinel框架用于控制迸发和流量，达到限流的效果，除了jmeter外拓展一种思路。

Jmeter常规测试

例如要一个查询接口，限制每秒访问10，可以使用jmeter，线程组持续时间1秒，循环次数1，线程数填写个11 12这样。然后观察结果树。

然后记住取样器错误后执行的动作，选择停止线程，这样就不会重复执行。如下

然后添加结果树，如果限流得以实现，那么结果树中，第11个接口返回将会提示限流的错误信息。

jmeter的测试应该适用于绝大多数接口了。

使用python协程测试

简单起见，考虑使用python代码完成，python有多进程，多线程，协程，多线程由于GIL的存在，实际是交替执行的，因此不考虑。由于这边要处理的是http请求，因此直接考虑协程+异步的组合了。

要求：同一时间只能同时有五个导入线程在执行，超出就报正在忙。

在同目录下有个文件夹，里面放了多个excel，前面分析过每个excel内容不能相同，于是只得出此下策，文件名有规律的设置方便代码里读取。（真实场景下也可以写个专门生成excel的脚本，然后再去测试接口）

import asyncio
import aiohttp
from aiohttp import FormData


class ImportOrder:
    def __init__(self):
        self.url = '' # 略
        self.data = {} # 略
        self.login_data = {"username": "", "password": ""}
        self.login_url = '' # 略

    async def request_and_get_result(self, n):
        filename = 'orderImport'+str(n)+'.xlsx'
        xlsx_path = 'xlsDir/'+filename

        data = FormData()
        data.add_field("params1", "")
        data.add_field("params2", "")
        data.add_field('file',
                       open(xlsx_path, 'rb'),
                       filename=filename,
                       content_type='application/vnd.ms-excel')

        async with aiohttp.ClientSession() as session:
              # 登陆一下cookieClientSession共享
            await session.post(self.login_url, data=self.login_data)  
            async with session.post(self.url, data=data) as response:
                return await response.text()

    # 协程+异步
    def async_aio(self):
        tasks = [asyncio.ensure_future(self.request_and_get_result(n)) for n in range(1, 8)]
        loop = asyncio.get_event_loop()
        loop.run_until_complete(asyncio.wait(tasks))
        for task in tasks:
            print(task.result())


if __name__ == '__main__':
    a = ImportOrder()
    a.async_aio()