Python实现HTTP并发压力测试脚本详解

方予

发布时间：2026-10-12 | 2人浏览过

原创

本文提供基于Python标准库实现的HTTP并发测试完整示例。通过调整thread_count变量控制最大线程数，利用_thread模块启动多线程执行GET请求。脚本实时统计错误次数并记录毫秒级起止时间，适用于快速验证服务器在高并发下的响应状态与稳定性，代码仅供参考学习。

通过修改变量thread_count的值，可以指定系统允许的最大并发数量，这直接对应于创建的线程总数。

测试任务执行完毕后，程序会打印输出统计到的失败请求次数，同时显示测试开始与结束的具体时间，时间单位精确到毫秒。

该示例主要用于帮助读者理解Python中的并发处理机制，内容仅供技术参考。

#!/usr/bin/python3

import sys, time, json, _thread
import http.client, urllib.parse

thread_count = 100  #并发数量
now_count = 0
error_count = 0
begin_time = ''

lock_obj = _thread.allocate()

def test_http_engine():
  global now_count
  global error_count
  global thread_count
  global begin_time
  conn = None
  if now_count == 0:
    begin_time = int(round(time.time() * 1000))
  try:
    conn = http.client.HTTPConnection("192.168.1.1", 80)
    conn.request('GET', '/')

    response = conn.getresponse()
    data = response.read()
    print (data)

    if json.dumps(response.status) != '200':
      error_count += 1;
      print ('error count: ' + str(error_count))

    sys.stdout.flush()
    now_count += 1
    if now_count == thread_count:
      print ('### error count: ' + str(error_count) + ' ###')
      print ('### begin time : ' + str(begin_time))
      print ('### end time  : ' + str(int(round(time.time() * 1000))))

  except Exception as e:
    print (e)
  finally:
    if conn:
      conn.close()

def test_thread_func():
  global now_count
  global lock_obj
  cnt = 0

  lock_obj.acquire()
  print ('')
  print ('=== Request: ' + str(now_count) + ' ===')

  cnt += 1
  test_http_engine()
  sys.stdout.flush()
  lock_obj.release()


def test_main():
  global thread_count
  for i in range(thread_count):
    _thread.start_new_thread(test_thread_func, ())

if __name__=='__main__':
  test_main()
  while True:
    time.sleep(5)