python异步高并发_通过python异步通讯方式构建高并发压力测试工具

背景说明

在工作中，要对一个接口进行压测，我当时就想通过python自己编写一个压力发生器。

初步方案（单线程循环发送）

通过循环向服务端发送请求，代码如下：

#采用单步循环的方式循环测试

import requests,time

def run(runnum):

url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"

for i in range(runnum):

str_res = requests.get(url)

if __name__ == "__main__":

start_time = time.time()

run(100)

end_time = time.time()

#print("循环次数：",str(counut))

print("开始时间：",str(start_time))

print("结束时间：",str(end_time))

print("运行时间：",str(end_time - start_time))

测试结果如下：

单线程

开始时间： 1536545804.5258229

结束时间： 1536545879.1070435

运行时间： 74.58122062683105

开始时间： 1536546027.6947124

结束时间： 1536546100.3893104

运行时间： 72.69459795951843

开始时间： 1536546205.2600951

结束时间： 1536546270.549498

运行时间： 65.28940296173096

开始时间： 1536546368.2361982

结束时间： 1536546435.7475684

运行时间： 67.51137018203735

开始时间： 1536546640.4913867

结束时间： 1536546712.5064435

运行时间： 72.015056848526

运行时间很长，对程序进行了分析，因为循环是单线程并且是同步的，发送请求后，必须等待收到响应，才会发送下一个请求，效率很低，并且循环对压力机的CPU资源消耗较大。

多线程方案

考虑通过多线程提高测试效率，代码如下：

import threading

import requests

import time

url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"

def run_thread(snum,enum):

for i in range(snum,enum):

#s = requests.session()

#关闭长连接

headers = {'Connection': 'close'}

str_res = requests.get(url,headers=headers)

#从发送请求到收到响应消耗的时间，单位微妙。

etime = str_res.elapsed.microseconds/1000000

threads = []

for i in range(0,10):

#循环生成线程

t = threading.Thread(target=run_thread, args=(0,10))

threads.append(t)

if __name__ == "__main__":

start_time = time.time()

for dd in range(0,10):

#启动线程

threads[dd].start()

for dd in range(0,10):

threads[dd].join()

end_time = time.time()

print("开始时间：",str(start_time))

print("结束时间：",str(end_time))

print("运行时间：",str(end_time - start_time))

测试结果如下：

单线程

多线程（10）

开始时间： 1536545804.5258229

结束时间： 1536545879.1070435

运行时间： 74.58122062683105

开始时间： 1536546926.9662883

结束时间： 1536546935.9247773

运行时间： 8.958488941192627

开始时间： 1536546027.6947124

结束时间： 1536546100.3893104

运行时间： 72.69459795951843

开始时间： 1536546962.368912

结束时间： 1536546970.9876196

运行时间： 8.618707656860352

开始时间： 1536546205.2600951

结束时间： 1536546270.549498

运行时间： 65.28940296173096

开始时间： 1536546990.911677

结束时间： 1536547001.0316806

运行时间： 10.120003700256348

开始时间： 1536546368.2361982

结束时间： 1536546435.7475684

运行时间： 67.51137018203735

开始时间： 1536547021.2030604

结束时间： 1536547030.7051418

运行时间： 9.502081394195557

开始时间： 1536546640.4913867

结束时间： 1536546712.5064435

运行时间： 72.015056848526

开始时间： 1536547046.3298163

结束时间： 1536547054.6956365

运行时间： 8.365820169448853

测试效率有很大提高，但也存在问题，当启动线程较多时，压力机资源消耗大，在同一个线程内部，还是同步进行，效率较低。

异步通讯方案

asyncio可以实现单线程并发IO操作。如果仅用在客户端，发挥的威力不大。如果把asyncio用在服务器端，例如Web服务器，由于HTTP连接就是IO操作，因此可以用单线程+coroutine实现多用户的高并发支持。

#采用异步通讯的方式发压

import requests,time

import asyncio

from aiohttp import ClientSession

import aiohttp

url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"

tasks = []

async def run():

async with ClientSession() as session:

async with session.get(url) as response:

response = await response.read()

#print(response)

if __name__ == "__main__":

loop = asyncio.get_event_loop()

start_time = time.time()

for i in range(100):

tasks.append(run())

end_time = time.time()

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

#print("循环次数：",str(counut))

print("开始时间：",str(start_time))

print("结束时间：",str(end_time))

print("运行时间：",str(end_time - start_time))

测试结果：

单线程

多线程（10）

异步

开始时间： 1536545804.5258229

结束时间： 1536545879.1070435

运行时间： 74.58122062683105

开始时间： 1536546926.9662883

结束时间： 1536546935.9247773

运行时间： 8.958488941192627

开始时间： 1536565502.732898

结束时间： 1536565502.732898

运行时间： 0.0

开始时间： 1536546027.6947124

结束时间： 1536546100.3893104

运行时间： 72.69459795951843

开始时间： 1536546962.368912

结束时间： 1536546970.9876196

运行时间： 8.618707656860352

开始时间： 1536565502.732898

结束时间： 1536565502.732898

运行时间： 0.0

开始时间： 1536546205.2600951

结束时间： 1536546270.549498

运行时间： 65.28940296173096

开始时间： 1536546990.911677

结束时间： 1536547001.0316806

运行时间： 10.120003700256348

开始时间： 1536565502.732898

结束时间： 1536565502.732898

运行时间： 0.0

开始时间： 1536546368.2361982

结束时间： 1536546435.7475684

运行时间： 67.51137018203735

开始时间： 1536547021.2030604

结束时间： 1536547030.7051418

运行时间： 9.502081394195557

开始时间： 1536565502.732898

结束时间： 1536565502.732898

运行时间： 0.0

开始时间： 1536546640.4913867

结束时间： 1536546712.5064435

运行时间： 72.015056848526

开始时间： 1536547046.3298163

结束时间： 1536547054.6956365

运行时间： 8.365820169448853

开始时间： 1536565502.732898

结束时间： 1536565502.732898

运行时间： 0.0

性能大大提高，但还有一个需要优化的地方，tasks采用的是list,如果数量多了，会占用大量内存，下步进行优化。

python异步高并发_通过python异步通讯方式构建高并发压力测试工具相关推荐

阿里云 mysql 高可用_如何在阿里云上构建高可用应用
原标题:如何在阿里云上构建高可用应用业务高可用是我们每个项目的需求,一个经常故障的项目,会让我们觉得不靠谱而选择放弃,从而导致项目的失败.今天,我们来聊一聊,如何让你自己的业务能够更加稳固的运行! ...
异步非阻塞_细说同步异步、阻塞非阻塞
同步.异步同步.异步分别指的是一种通讯方式,当 cpu 不需要执行线程上下文切换就能完成任务,此时便认为这种通讯方式是同步的,相对的如果存在cpu 上下文切换,这种方式便是异步. 这里通过一个去食堂 ...
go python php 压力测试_pyLot 基于python的压力测试工具
因为组内的产品是一个供上海甚至全球所有designer使用的网站,所以并发性很高,于是一直存在访问速度不佳的诟病.为了提高访问速度,就需要对该网站的性能进行测试.于是在网上查了并发性测试和自动化测试的 ...
python的web压力测试工具-pylot安装使用
pylot是python编写的一款web压力测试工具.使用比较简单.而且测试结果相对稳定. 这里不得不鄙视一下apache 的ab测试,那结果真是让人蛋疼,同样的url,测试结果飘忽不定,看得人心惊肉 ...
高并发ab压力测试工具
Apache ab 介绍 : Apache Benchmark(简称ab) 是Apache安装包中自带的压力测试工具注: centos6 默认安装 ab 安装命令 yum install y htt ...
网站测压机器人_压力测试工具(bots)
压力测试工具(bots) bots机器人介绍: bots就是机器人,可以模拟客户端的连接.通讯.操作等,是一种对服务端进行模拟测试的轻量级客户端程序,并且bots没有渲染等部分.引擎提供了Python ...
html5 并发测试工具,超实用压力测试工具－ab工具
吞吐率(Requests per second) 概念:服务器并发处理能力的量化描述,单位是reqs/s,指的是某个并发用户数下单位时间内处理的请求数.某个并发用户数下单位时间内能处理的最大请求数,称 ...
windows linux 并发,Windows/Linux安装压力测试工具siege
Siege是一款高性能的Http压力测试工具. Siege支持身份验证.cookies.http.https和ftp协议. Linux安装Siege 如果要支持https,需要先下载安装openssl ...
python访问数据库如何解决高并发_使用 Python 和 Oracle 数据库实现高并发性
随着趋势发展的核心转向更多而不是更快发展,最大限度地提高并发性的重要性日益凸显.并发性使得编程模式发生了新的转变,可以编写异步代码,从而将多个任务分散到一组线程或进程中并行工作.如果您不是编程新手并且 ...

python异步高并发_通过python异步通讯方式构建高并发压力测试工具

python异步高并发_通过python异步通讯方式构建高并发压力测试工具相关推荐

最新文章

热门文章