背景：使用python进行并发请求，进行压力测试
#!/usr/bin/env python2
# -*- coding: utf-8 -*-import threading
import urllib2
import time
from Queue import Queue# 配置参数
url = "http://192.xxx10" # 替换为你要测试的URL
num_requests = 10000000 # 总请求数
concurrent_threads = 400 # 并发线程数# 记录成功和失败的请求数
success_count = 0
failure_count = 0
lock = threading.Lock()def send_request(q):global success_count, failure_countwhile not q.empty():try:response = urllib2.urlopen(url, timeout=5)if response.getcode() == 200:with lock:success_count += 1qps = success_count % 100if qps == 0:print success_countelse:with lock:failure_count += 1except Exception as e:with lock:failure_count += 1finally:q.task_done()if __name__ == "__main__":print "Starting load test on %s" % urlstart_time = time.time()# 创建队列并填充任务q = Queue(maxsize=num_requests)for _ in range(num_requests):q.put(1)# 创建并启动线程threads = []for i in range(concurrent_threads):t = threading.Thread(target=send_request, args=(q,))t.daemon = Truet.start()threads.append(t)# 等待所有任务完成q.join()# 结束时间end_time = time.time()elapsed_time = end_time - start_timeprint "\n--- Load Test Results ---"print "Total Requests: %d" % num_requestsprint "Concurrency Level: %d" % concurrent_threadsprint "Successful Requests: %d" % success_countprint "Failed Requests: %d" % failure_countprint "Time taken for tests: %.2f seconds" % elapsed_timeprint "Requests per second: %.2f" % (num_requests / elapsed_time)
