[Python] 多线程 HTTP/HTTPS 访问延迟测试工具

作者 huhamhire,暂无评论,2014年2月5日 13:14 程序实践

由于 hosts 中的内容绝大部分都是指向 HTTP/HTTPS 服务,为了测试实际的访问效果,在做完之前的 ICMP 延迟测试之后,我准备继续做一遍类似 ping 的 http 访问延迟测试,并记录页面访问的状态信息。当然,这次实现的依然只是最后成品模块的原型。

在目前总共数千条的域名规模的情况下,在各域名与其对应的 IP 组合后,总共有将近 35k+ 个组合,需要在 HTTP 测试的时候逐一进行测试。又由于每个组合需要分别对 HTTP 以及 HTTPS 连接进行测试,即使每项测试如果同 Ping 测试一样只统计 4 次结果的话,整体的测试规模也已经达到了 28 万次的级别。

对于在单点进行的网络测试而言,这样的测试规模已经算比较大的了,唯一能做的就是提高线程并发数量。不过网络并发测试基本受制于网络带宽限制,事实上我在只使用 512 个线程并发的状态下,峰值即可占满家里 20Mb/s 宽带的最大带宽。所以,在网络一般的情况下,这项测试注定是一个漫长的过程。

虽然如此,我还是觉得使用多进程并发对于解决这个问题帮助不大,毕竟瓶颈依旧在于网络这部分。所以,我仍旧采用了与之前类似的多线程控制结构,首先是一个用于测试 http 以及 https 连接延迟的类。


  1 class HTTPTest(threading.Thread):
  2     STATUS_DESC = {
  3         200: "OK",
  4         204: "No Content",
  5         301: "Moved",
  6         302: "Redirect",
  7         400: "Bad Request",
  8         403: "Forbidden",
  9         404: "Not Found",
 10         408: "Timed Out",
 11         500: "Server Error",
 12         502: "Bad Gateway",
 13         503: "Unavailable",
 14         600: "Unknown",
 15         601: "Conn Reset"
 16     }
 17     url = ""
 18     conn = None
 19     http_stat = {}
 20     _response_log = {}
 21 
 22     def __init__(self, ip, domain, comb_id, results, semaphore,
 23                  req_count=4, timeout=5):
 24         threading.Thread.__init__(self)
 25         self._ip = ip
 26         self._domain = domain
 27         self._comb_id = comb_id
 28         self.results = results
 29         self.sem = semaphore
 30         self.req_count = req_count
 31         self.timeout = timeout
 32 
 33     def _set_http_req(self):
 34         self.url = "http://%s/" % self._domain
 35         self.conn = httplib.HTTPConnection(host=self._ip,
 36                                            timeout=self.timeout)
 37 
 38     def _set_https_req(self):
 39         self.url = "https://%s/" % self._domain
 40         self.conn = httplib.HTTPSConnection(host=self._ip,
 41                                             timeout=self.timeout)
 42 
 43     def _check(self):
 44         start_time = time.time()
 45         try:
 46             self.conn.request(method="GET", url=self.url)
 47             response = self.conn.getresponse()
 48         except socket.timeout:
 49             return 408, None
 50         except socket.error, e:
 51             if len(e.args) == 1:
 52                 _err_no, _err_msg = 'UNKNOWN', e.args[0]
 53                 if "timed out" in _err_msg:
 54                     return 408, None
 55                 else:
 56                     sys.stderr.write("\r  %s - %s: %s\n" %
 57                                      (self.url, self._ip, _err_msg))
 58             else:
 59                 _err_no, _err_msg = e.args
 60 
 61             if _err_no == 10054:
 62                 return 601, None
 63             else:
 64                 return 600, None
 65         except httplib.BadStatusLine:
 66             return 600, None
 67         finally:
 68             self.conn.close()
 69         delay = time.time() - start_time
 70         return response.status, delay
 71 
 72     def session(self):
 73         response = {}
 74         try:
 75             for method in ["http", "https"]:
 76                 status_log = []
 77                 delay_log = []
 78                 for i in range(self.req_count):
 79                     exec ("self._set_%s_req()" % method)
 80                     status, delay = self._check()
 81                     if status not in status_log:
 82                         status_log.append(status)
 83                     delay_log.append(delay)
 84                 response[method] = {"status": status_log,
 85                                     "delay": delay_log}
 86                 self.show_state(status_log)
 87             self._response_log = response
 88         finally:
 89             self.sem.release()
 90 
 91     def _stat_delay(self, delay_log):
 92         log = [delay * 1000 for delay in delay_log if delay is not None]
 93         if log:
 94             min_delay = round(min(log), 3)
 95             max_delay = round(max(log), 3)
 96             avg_delay = round(sum(log) / len(log), 3)
 97             loss = round(1.0 * len(log) / self.req_count, 3)
 98             return {"min": min_delay, "max": max_delay,
 99                     "avg": avg_delay, "ratio": loss}
100         else:
101             return {"min": None, "max": None,
102                     "avg": None, "ratio": 0}
103 
104     def _stat_status(self, status_log):
105         return "|".join([str(status) for status in sorted(status_log)])
106 
107     def stat(self):
108         response_log = self._response_log
109         stat = {
110             "ip": self._ip,
111             "domain": self._domain,
112             "req_count": self.req_count
113         }
114         for method, log in response_log.iteritems():
115             stat[method] = {"delay": self._stat_delay(log["delay"]),
116                             "status": self._stat_status(log["status"])}
117         self.http_stat = stat
118 
119     def set_results(self):
120         self.results[self._comb_id] = self.http_stat
121 
122     def show_state(self, status_log):
123         msg = "HTTP: %s - %s" % (self.url, self._ip)
124         if status_log:
125             status_flag = min(status_log)
126             if status_flag == 200:
127                 Progress.show_status(msg, self.STATUS_DESC[status_flag])
128             elif status_flag in self.STATUS_DESC.keys():
129                 Progress.show_status(msg, self.STATUS_DESC[status_flag], 1)
130             else:
131                 Progress.show_status(msg, str(status_flag), 1)
132         else:
133             Progress.show_status(msg, "NO STATUS", 1)
134         Progress.progress_bar()
135 
136     def run(self):
137         self.session()
138         self.stat()
139         self.set_results()

整个 HTTP 延迟测试的原理很简单,在向指定的服务器 IP 地址发送完针对相关域名页面的请求以后,记录从发送请求到接收到返回数据的时间即可。如果出现访问的异常状态,一一进行相应的处理并记录。由于 Python 提供了 httplib 可以很方便的实现这里的需求,所以这里的代码只需要处理上层功能即可,写起来比较简单。

说到异常处理部分,我主要使用了常规的 HTTP 错误代码来记录相应的异常,不过由于 HTTP 错误代码并没有对于链接被重置这样的典型被墙状态,所以我就自己随便定义了 600 - “未知”,601 - “链接被重置”这两个错误用于记录。

接下来,同之前的测试一样,给出多线程的控制部分。与之前的两篇文章一样,这一部分没有太多的新东西,就不详细说明了。


 1 class MultiHTTPTest(object):
 2     # Limit the number of concurrent sessions
 3     sem = threading.Semaphore(0x200)
 4 
 5     def __init__(self, combinations):
 6         self.combs = combinations
 7         self._responses = {}
 8 
 9     def http_test(self):
10         Progress.set_total(len(self.combs))
11         Progress.set_counter(self._responses)
12         threads = []
13         for comb in self.combs:
14             self.sem.acquire()
15             http_test_item = HTTPTest(comb["ip"], comb["domain"], comb["id"],
16                                       self._responses, self.sem)
17             http_test_item.start()
18             threads.append(http_test_item)
19 
20         for http_test_item in threads:
21             http_test_item.join()
22 
23         Progress.progress_bar()
24         return self._responses

下面是测试的主程序部分。因为完整的测试规模较大,在我目前的网络环境下,完成整个测试需要数十分钟时间,以后尽量会找办法做进一步的优化。


1 if __name__ == '__main__':
2     SourceData.connect_db()
3     combs = SourceData.get_http_test_comb()
4 
5     http_tests = MultiHTTPTest(combs)
6     results = http_tests.http_test()
7 
8     SourceData.set_multi_http_test_dict(results)

最后,照例给一下我跑 HTTP 测试的截图。

http
关键词:PPTP , 工具 DIY , 网络测试
登录后进行评论