性能压测的时候，随着并发压力的增加，系统响应时间和吞吐量如何变化，为什么？
性能测试通常分为三个阶段，分别是预期目标测试、负载测试和压力测试。如上图所示，黄色区间代表预期目标测试，绿色区间代表负载测试，红色区间代表压力测试，这三个测试阶段的并发压力是依次递增的。
可以看到，在预期目标测试期间，系统完全可以承担负载压力，因此随着负载的增高，系统所能处理的并发请求数（绿色线）也呈线性增长趋势，直至达到系统预期的性能目标。在这一阶段，系统的负载压力较小，可以快速响应所有并发请求，因此响应时间（蓝色线）很短，并且在整个阶段几乎没有什么变化。
考虑到吞吐量 = (1000 / 响应时间ms) x 并发数这一公式，可以得知在这一阶段吞吐量也是近似于线性增长的。
当系统达到预期性能指标后，并发压力继续增大，这时进入到负载测试阶段，系统资源的负载进一步升高，对请求的响应时间也开始逐渐增加，相应的，系统能够承受的并发数的增长速度便出现明显减缓的趋势。越接近系统资源负载的安全临界点，则请求响应时间越长，并发数增长越慢。这一阶段系统吞吐量将出现先增后降的情况，即一开始并发数增长率比响应时间增长率高的时候吞吐量是呈增长趋势的，而当响应时间增长率超过并发数增长率，这时候吞吐量便开始下降。
继续增加并发压力，此时系统已经满载甚至出现崩溃，越来越多的请求无法被及时处理，因此请求响应时间呈指数级快速增长，而系统的并发数则呈现出明显的下降趋势，直至最后整个系统崩溃，无法响应任何请求。这一阶段吞吐量也将呈现显著的下降趋势。
用你熟悉的编程语言写一个 web 性能压测工具，输入参数：URL，请求总次数，并发数。输出参数：平均响应时间，95% 响应时间。用这个测试工具以 10 并发、100 次请求压测 www.baidu.com。
# coding: utf-8
import socket
from threading import Thread
import time
from urllib import urlopen
TIMEOUT = 2
def average(_list):
sum = 0
for i in _list:
sum += i
return sum/len(_list)
def percent_result(_list, percent):
index = int(len(_list)*percent/100.0)
return _list[index-1]
class RequestThread(Thread):
def __init__(self, url):
self.url = url
self.response_time = -1
self.response_status = False
super(RequestThread, self).__init__()
def run(self):
try:
start_time = time.clock()
response = urlopen(self.url)
self.response_time = time.clock() - start_time
except:
pass
self.response_status = True
class WebPerformanceTester(object):
def __init__(self, timeout):
self._threads = []
self.results = []
self._timeout = timeout
socket.setdefaulttimeout(timeout)
@property
def timeout(self):
return self._timeout
@timeout.setter
def timeout(self, timeout):
self._timeout = timeout
socket.setdefaulttimeout(timeout)
def test(self, url, requests_count, concurrency_count):
assert isinstance(requests_count, int)
assert isinstance(concurrency_count, int)
assert concurrency_count != 0
print '性能测试请求地址：{}'.format(url)
print '总请求数：{} 并发数：{} 请求超时时间：{}秒'.format(
requests_count, concurrency_count, self.timeout)
print '正在测试......'
self.results = []
turn = requests_count / concurrency_count
remainder = requests_count % concurrency_count
for _ in range(turn):
self._prepare_threads(url, concurrency_count)
self._do_testing()
self._show_result(requests_count)
def _clear_threads_pool(self):
self._threads = []
def _prepare_threads(self, url, concurrency_count):
self._clear_threads_pool()
for _ in range(concurrency_count):
self._threads.append(RequestThread(url))
def _do_testing(self):
for _thread in self._threads:
_thread.start()
while True:
if all([_thread.response_status for _thread in self._threads]):
break
time.sleep(0.5)
for _thread in self._threads:
self.results.append(_thread.response_time)
def _show_result(self, requests_count):
print '测试结束'
self.results.sort()
self.results = filter(lambda x: x>0, self.results)
print '有效响应数：{} 超时响应数：{}'.format(
len(self.results), requests_count - len(self.results))
print '最小响应时间：{}ms 最大响应时间:{}ms'.format(
self.results[0]*1000, self.results[-1]*1000)
print '平均响应时间：{}ms 95%响应时间：{}ms'.format(
average(self.results)*1000, percent_result(self.results, 95)*1000)
def main():
tester = WebPerformanceTester(TIMEOUT)
tester.test('http://www.baidu.com', 100, 10)
main()
运行结果：
性能测试请求地址：http://www.baidu.com
总请求数：100 并发数：10 请求超时时间：2秒
正在测试......
测试结束
有效响应数：100 超时响应数：0
最小响应时间：14.9305ms 最大响应时间:188.9546ms
平均响应时间：81.886768ms 95%响应时间：136.3102ms
网友评论