项目介绍
pyWebTest是一个性能和负载测试工具，采用纯python编写，可以并发地对目标网站生成请求，测试响应时间和吞吐量，并生成完整的测试报告。
通过简单地修改测试部分的代码，可以很快的实现对远程API的压力测试，事实上，可以通过编写测试脚本，实现很多复杂的测试计划。
测试报告通过标准输出给出。(文件和网页形式的报告功能没有实现)
使用说明
下载和安装
pyWebTest可以通过Git下载完整的源代码，并且不需要额外的设置就可以直接运行。
$ git clone git@github.com:saukymo/pyWebTest.git
$ cd pyWebTest
$ python main.py -h
使用方法
通过-h参数可以看到简单的参数介绍，一共有4个可选参数，-a是并发进程数，默认模拟两个用户请求网；，-d是测试时间，超过时间测试中止，设定时间之后的请求会被忽略，默认持续10秒；-i是统计时对数据采用的分段数，默认将整个测试分成10段统计；-r是启动进程的时间，所有的并发将在启动时间内均匀的开始，平稳地增加，启动时间到达后，达到预先设定的并发进程数，默认为0秒，即一开始就达到最大负载进行测试。-u是测试的目标网站，不能省略。
$ python main.py -h
usage: main.py [-h] [-a [A]] [-d [D]] [-i [I]] [-r [R]] -u U
optional arguments:
-h, --help show this help message and exit
-a [A] number of agents
-d [D] test duration in seconds
-i [I] number of time-seriels interval
-r [R] rampup in seconds
-u U test target url
测试和报告样例
样例测试选择百度作为测试对象，一共产生20个用户模拟请求，测试一共持续30秒，统计分成10段，即3秒统计分析一次。启动时间10秒，即每0.5秒增加一个新的用户。测试的报告如下：
python main.py -a 20 -d 30 -i 10 -r 10 -u http://www.baidu.com
===Summary===
transactions: 1788 hits
errors: 0
runtime: 30 secs
rampup: 10 secs
time-series interval: 3.000 secs
test start: 2016-07-19 00:18:13
test finish: 2016-07-19 00:18:43
===All Transactions===
Response Time Summary (secs):
count avg min 50pct 80pct 90pct max stdev
1773 0.283 0.123 0.293 0.351 0.396 0.509 0.076
Interval Details (secs):
interval count avg min 50pct 80pct 90pct max stdev
0 65 0.155 0.123 0.160 0.167 0.169 0.191 0.015
1 171 0.164 0.124 0.163 0.173 0.193 0.254 0.022
2 194 0.230 0.124 0.245 0.252 0.258 0.349 0.040
3 192 0.306 0.222 0.299 0.352 0.400 0.458 0.055
4 196 0.308 0.231 0.299 0.352 0.396 0.509 0.056
5 193 0.313 0.242 0.300 0.354 0.399 0.452 0.054
6 190 0.312 0.242 0.300 0.352 0.399 0.506 0.058
7 190 0.314 0.243 0.302 0.355 0.402 0.509 0.058
8 192 0.313 0.240 0.301 0.357 0.403 0.506 0.061
9 190 0.316 0.241 0.299 0.359 0.405 0.508 0.069
===Agent Details===
Agent Starttime Requests Errors Bytes received Avg Response Time(secs) Avg throughput(req/sec)
0 0.00 118 0 11705655 0.256 3.904
1 0.50 113 0 11211349 0.263 3.801
2 1.00 112 0 11110130 0.259 3.859
3 1.50 109 0 10812108 0.263 3.805
4 2.00 107 0 10613035 0.263 3.802
5 2.50 100 0 9920078 0.276 3.627
6 3.00 100 0 9920273 0.272 3.680
7 3.51 95 0 9425688 0.281 3.555
8 4.01 94 0 9324440 0.278 3.599
9 4.51 89 0 8828827 0.287 3.488
10 5.01 86 0 8531848 0.291 3.438
11 5.51 82 0 8133568 0.299 3.345
12 6.01 83 0 8233050 0.290 3.442
13 6.51 79 0 7837821 0.298 3.359
14 7.01 76 0 7538452 0.306 3.270
15 7.51 75 0 7438603 0.302 3.314
16 8.01 70 0 6944360 0.316 3.165
17 8.51 72 0 7141987 0.301 3.316
18 9.01 67 0 6645907 0.314 3.181
19 9.51 66 0 6547565 0.311 3.217
项目模块和实现
运行参数分析
参数分析采用python的argparse模块，通过这个模块，可以很简单的增加可以设置的参数，可以设置参数类型，并且自动生成帮助文档。代码如下：
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('-a', nargs='?', help='number of agents', default=2)
arg_parser.add_argument('-d', nargs='?', help='test duration in seconds', default=10)
arg_parser.add_argument('-i', nargs='?', help='number of time-seriels interval', default=10)
arg_parser.add_argument('-r', nargs='?', help='rampup in seconds', default=0.0)
arg_parser.add_argument('-u', help='test target url', required=True)
args = vars(arg_parser.parse_args())
args就是一个包含所有参数的字典了，例如通过args.get('i')就可以获取用户运行时设置的参数i了。
进程间通信
虽然经过测试发现，多线程与多进程版本差别不大，所以最后改用多线程完成了整个项目，但在一开始还是使用的多进程版本。那么每个进程进行测试的结果就需要通过进程间通信的方式共享出来。这里采用生产者消费者模式的Queue来实现。
首先需要定义一个QueueReader类来充当消费者的形式，并在测试开始前开启这个进程，并且设置daemon属性为True，这样这个进程就会随着主进程结束一起结束。
queue = multiprocessing.Queue()
queue_reader = QueueReader(queue)
queue_reader.daemon = True
queue_reader.start()
在每个测试进程中，将结果通过Queue.put方法将结果放到队列中。
self.queue.put(fields)
队列满时，会产生Queue.Full错误，测试进程挂起，等待队列被消耗，队列空时，产生Queue.Empty错误，读取进程等待一小段时间(这里设置为0.05秒)
测试进程管理
首先，将每个进程的预先设置好，然后启动一个全局的计时器，之后再依次开始每个进程的测试。
threads = []
for idx in range(self.num_threads):
agent = Agent(self.queue, idx, self.run_time)
agent.daemon = True
threads.append(agent)
for idx, agent in enumerate(threads):
spacing = 1.0 * self.rampup / self.num_threads
if idx > 0:
time.sleep(spacing)
agent.start()
for agent in threads:
agent.join()
其中，rampup的功能是在第二个循环中实现的，首先计算每个进程的平均启动时间，然后等待足够的时间后再启动一个新的进程即可。
测试进程实现
测试进程的部分其实很简单，其中真正的请求部分只有三行：
connect = urllib2.urlopen(url)
content = connect.read()
resp_data = len(content)
通过修改这个部分的代码，可以实现对API的请求测试，也可以实现各种复杂的测试计划。
整个测试进程的框架如下：
elapsed_time = 0
while elapsed_time < self.run_time:
error = ''
start = time.time()
try:
# 测试部分代码
except Exception, e:
error = str(e)
run_time = time.time() - start
elapsed_time = time.time() - UNION_START_TIME
# 测试结果field统计，然后将结果放到Queue中
Queue.put(field)
测试结果的统计
统计分两个部分进行。一个是从每个进程的角度分析，统计每个进程的平均响应时间和吞吐量。这个在每个进程完成所有测试之后统计即可，在设置了rampup参数时，这个统计可以看出测试机器的性能对结果的影响，例如在进程较少时，平均响应时间较短，进程较多时，平均响应时间较长。
另外一个则是从时间分段的角度进行分析，统计不同阶段的响应时间及其分布情况，这个需要记录所有的测试结果之后，根据测试的时间分组统计。这个统计可以看出目标网站的性能情况，看出不同阶段，请求进程数不同时，目标网站的响应时间变化。
由于这部分代码零散的分布在测试的各个阶段，所以就不贴代码了。
如果目标网站的响应时间较长，而测试时间较短，可能出现某个时间成功响应次数为0的情况，这个时候统计函数会出错，所以需要给它们都加上一个特殊判断，这里采用装饰器的方法判断：
def validator(func):
def _func(*args, **kargs):
if len(args[0]) == 0:
return 0.0
return func(*args, **kargs)
return _func
@validator
def minarg(seq):
return min(seq)
@validator
def maxarg(seq):
return max(seq)
@validator
def average(seq):
avg = (float(sum(seq)) / len(seq))
return avg
@validator
def standard_dev(seq):
avg = average(seq)
sdsq = sum([(i - avg) ** 2 for i in seq])
try:
stdev = (sdsq / (len(seq) - 1)) ** .5
except ZeroDivisionError:
stdev = 0
return stdev
@validator
def percentile(seq, percentile):
i = int(len(seq) * (percentile / 100.0))
seq.sort()
return seq[i]
测试报告生成
由于没有简单的重定向方法，新增加一个报告生成的方式需要大量重复代码，所以只完成了标准输出的方式。
报告分为三个部分：
- 测试总体概览，包括测试的时间、参数等等；
- 时间分段的响应时间统计；
- 测试进程的响应时间统计；
输出部分的代码如下：
print "\n===Summary==="
print "transactions: %d hits" % (queue_reader.trans_count)
print "errors: %d" % (queue_reader.error_count)
print "runtime: %d secs" % (test_duration)
print "rampup: %d secs" % (int(args.get('r')))
print "time-series interval: %0.3f secs" % (interval_time)
print ""
print "test start: %s" % (datetime.fromtimestamp(test_start_timestamp).strftime("%Y-%m-%d %H:%M:%S"))
print "test finish: %s" % (datetime.fromtimestamp(test_end_timestamp).strftime("%Y-%m-%d %H:%M:%S"))
print "\n===All Transactions==="
print "Response Time Summary (secs):"
print "%7s\t%7s\t%7s\t%7s\t%7s\t%7s\t%7s\t%7s" % ("count", "avg", "min", "50pct", "80pct", "90pct", "max", "stdev")
print "%7d\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f\n" % (len(all_run_time), average(all_run_time), minarg(all_run_time), percentile(all_run_time, 50),\
percentile(all_run_time, 80), percentile(all_run_time, 90), maxarg(all_run_time), standard_dev(all_run_time))
print "Interval Details (secs):"
print "%8s\t%7s\t%7s\t%7s\t%7s\t%7s\t%7s\t%7s\t%7s" % ("interval", "count", "avg", "min", "50pct", "80pct", "90pct", "max", "stdev")
for interval_num in range(interval):
interval_list = interval_run_time[interval_num]
print "%8d\t%7d\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f\t%7.3f" % (interval_num, len(interval_list), average(interval_list), minarg(interval_list), percentile(interval_list, 50),\
percentile(interval_list, 80), percentile(interval_list, 90), maxarg(interval_list), standard_dev(interval_list))
print "\n===Agent Details==="
print "Agent\tStarttime\tRequests\tErrors\tBytes received\tAvg Response Time(secs)\tAvg throughput(req/sec)"
for agent_detail in ag.details:
print agent_detail
性能优化和提速
Python本身的效率不是很高，这一点有可能会对测试结果产生影响。另外，由于默认的Python解释器存在GIL的问题，在多线程的情况下会出现严重的性能问题。所以需要一些方法来提高整个测试程序的运行速度。
需要注意的是，默认的测试代码比较简单，对计算能力要求不高，属于IO密集型的任务，任务的瓶颈在于网速，所以要求不高时，默认的Python完全能够胜任。
采用PyPy提速
PyPy是一个独立的解析器， 通过即时编译(JIT,Just-in-time)代码避免逐行解释执行来提升运行速度。由于我们的项目采用纯Python编写实现，所以可以简单的通过替换使用PyPy的解释器来达到提速的效果。
首先下载最新版本的PyPy：
wget https://bitbucket.org/pypy/pypy/downloads/pypy2-v5.3.1-linux64.tar.bz2
解压后，用新的解释器来执行程序即可。
/path/to/pypy/bin/pypy main.py -h
但是这样仍然存在GIL的问题，可以通过使用pypy-stm版本来解决这个问题，安装和使用的方法与普通版的一样。
不同版本的解释器测试结果比较
测试目标网站还是www.baidu.com。测试参数为20个并发进程，持续30秒，无启动时间。统计分为10段。测试使用的命令如下：
$ python main.py -a 20 -d 30 -i 10 -u http://www.baidu.com
$ pypy-stm/bin/pypy-stm main.py -a 20 -d 30 -i 10 -u http://www.baidu.com
$ pypy/bin/pypy main.py -a 10 -d 30 -i 10 -u http://www.baidu.com
在测试时间内，三个版本分别完成的请求数为1938，2033，1828次。
三个版本相差不多，而且由于测试时间较短，而新建进程开销较大，所以pypy的速度甚至慢于原版Python，而pypy-stm版本速度最快。
存在的问题
这个程序在我的Macbook pro上的结果比较奇怪，当进程数增加时，平均响应时间会显著减少，多次测试和分析之后发现应该是性能原因导致的，但是系统资源并没有被完全利用。而同样的代码在阿里云的Ubuntu服务器上运行正常，猜测是由于系统的多进程API差异导致的。目前没有发现解决方法。