Django并发模型测试

测试场景，10线程测试100请求的并发性能。

压测代码：

import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from statistics import mean

URL = "http://127.0.0.1:8000/api/test/"
TOTAL_REQUESTS = 100      # 总请求数
THREADS = 10             # 线程数
TIMEOUT = 20             # 超时（秒）

session = requests.Session()

def request_once():
    start = time.perf_counter()
    try:
        r = session.get(URL, timeout=TIMEOUT)
        cost = time.perf_counter() - start
        return r.status_code, cost
    except Exception:
        return "ERROR", None

def main():
    results = []
    start_time = time.perf_counter()
    
    with ThreadPoolExecutor(max_workers=THREADS) as executor:
        futures = [executor.submit(request_once) for _ in range(TOTAL_REQUESTS)]
        for f in as_completed(futures):
            results.append(f.result())
    
    total_time = time.perf_counter() - start_time
    success = [r for r in results if r[0] == 200]
    failed = TOTAL_REQUESTS - len(success)
    latencies = [r[1] for r in success]
    
    print("====== 压测结果 ======")
    print(f"URL: {URL}")
    print(f"总请求数: {TOTAL_REQUESTS}")
    print(f"线程数: {THREADS}")
    print(f"成功请求: {len(success)}")
    print(f"失败请求: {failed}")
    print(f"总耗时: {total_time:.2f}s")
    print(f"QPS: {TOTAL_REQUESTS / total_time:.2f}")
    
    if latencies:
        print(f"平均响应时间: {mean(latencies)*1000:.2f} ms")
        print(f"最大响应时间: {max(latencies)*1000:.2f} ms")

if __name__ == "__main__":
    main()
    

测试结果