一、Python 压测脚本

使用threading多线程模拟并发读写，打印详细耗时、成功率、QPS日志，支持自定义并发数、请求总量、读写比例。

依赖：redis-py

pipinstallredis

1.1、redis_concurrent_test.py

importthreadingimporttimeimportredisfromtypingimportDict# ===================== 配置区 =====================REDIS_HOST="127.0.0.1"REDIS_PORT=6379REDIS_PASSWORD=NoneREDIS_DB=0CONCURRENT_THREADS=20# 并发线程数TOTAL_REQUESTS=10000# 总请求次数TEST_KEY_PREFIX="test_concurrent:"# 读写比例 True=写, False=读；调整读写混合场景WRITE_RATIO=0.3# ==================================================# 全局统计stat:Dict={"success":0,"fail":0,"total_cost":0.0,"lock":threading.Lock()}defcreate_redis_client()->redis.Redis:"""创建redis连接（每个线程独立client，避免线程安全问题）"""returnredis.Redis(host=REDIS_HOST,port=REDIS_PORT,password=REDIS_PASSWORD,db=REDIS_DB,decode_responses=True,socket_timeout=5)defworker(thread_id:int,req_per_thread:int):"""单个线程工作函数"""r=create_redis_client()thread_start=time.perf_counter()print(f"[线程{thread_id:02d}] 启动，计划执行{req_per_thread}次请求")foriinrange(req_per_thread):key=f"{TEST_KEY_PREFIX}{thread_id}_{i}"start=time.perf_counter()try:ifi/req_per_thread<WRITE_RATIO:# 写操作 setr.setex(key,300,f"value_{thread_id}_{i}_{time.time()}")else:# 读操作 getr.get(key)cost=time.perf_counter()-startwithstat["lock"]:stat["success"]+=1stat["total_cost"]+=costexceptExceptionase:withstat["lock"]:stat["fail"]+=1print(f"[线程{thread_id:02d}] 请求异常 key={key}err={str(e)}")thread_cost=time.perf_counter()-thread_startprint(f"[线程{thread_id:02d}] 执行完成，耗时={thread_cost:.3f}s")defmain():print("="*60)print(f"Redis单机并发压测启动")print(f"地址:{REDIS_HOST}:{REDIS_PORT}| 并发线程:{CONCURRENT_THREADS}")print(f"总请求量:{TOTAL_REQUESTS}| 写比例:{WRITE_RATIO:.1%}")print("="*60)# 均分请求到各个线程base_req=TOTAL_REQUESTS//CONCURRENT_THREADS threads=[]start_time=time.perf_counter()fortidinrange(CONCURRENT_THREADS):t=threading.Thread(target=worker,args=(tid,base_req))threads.append(t)t.start()# 等待所有线程结束fortinthreads:t.join()total_time=time.perf_counter()-start_time# 输出汇总统计print("\n"+"="*60)print("【压测结果汇总】")print(f"总耗时:{total_time:.3f}s")print(f"成功请求:{stat['success']}")print(f"失败请求:{stat['fail']}")ifstat["success"]>0:avg_latency_ms=(stat["total_cost"]/stat["success"])*1000qps=stat["success"]/total_timeprint(f"平均延迟:{avg_latency_ms:.3f}ms")print(f"QPS(成功):{qps:.2f}req/s")print("="*60)if__name__=="__main__":main()

1.2、关键说明

1. 线程安全：每个线程新建独立Redis实例，禁止多线程共用同一个client（redis-py同步客户端非线程安全）
2. 支持读写混合压测，修改WRITE_RATIO：
   • 1.0纯写压测
   • 0.0纯读压测
3. 日志输出：
   • 每个线程启动/结束日志
   • 异常请求打印错误信息
   • 最终汇总：总耗时、成功失败数、平均延迟、QPS
4. 适用于单机Redis；如需集群压测只需修改连接初始化

1.3、进阶优化建议（可选）

如果想要更高压测性能：

1. 使用gevent协程版本（大量并发场景比thread高效）
2. 开启redis pipeline批量请求
3. 使用hiredis加速解析：pip install hiredis，客户端自动启用

1.4、协程版本补充（高并发推荐）

略过

二、redis内部提供的命令

2.1、命令 redis-benchmark

命令例子：进入到redis的安装目录执行 redis-benchmark -a 123456 -c 100 -n 100000 -t get

如果是docker方式： docker exec -it redis redis-benchmark -a 123456 -c 100 -n 100000 -t get

2.2、参数解释

-a 123456：Redis 认证密码 -c 100：并发连接数 -n 100000：总请求次数 -t get：只压测 GET 命令

2.3、压测结果解读（GET 纯读，100 并发，10 万请求）

throughput summary:104058.27requests per second latency summary(msec): avg min p50 p95 p99 max0.4960.1760.4390.8151.1914.807

核心指标

QPS ≈ 10.4 万 /s Docker 内单机 Redis 纯读，这个性能属于非常优秀。 延迟分布 平均延迟：0.496ms p50：0.439ms（一半请求低于该值） p95：0.815ms p99：1.191ms 最大延迟：4.807ms（少量毛刺）

2.4、简单分析

✅ 绝大部分请求稳定在 1ms 以内；
✅ 长尾延迟不高，没有严重阻塞；
⚠️ 注意：redis-benchmark 测试的是预设 key，不存在 key 缺失、大 value、复杂命令；真实业务 QPS 会低于该数值。

2.5、建议继续测试（复制直接运行）

1）纯 SET 写压测（最有参考价值）

docker exec -it redis redis-benchmark -a 123456 -c 100 -n 100000 -t set

2）读写混合（默认场景 SET/GET 等多条命令）

docker exec -it redis redis-benchmark -a 123456 -c 100 -n 100000

3）调高并发到 200，观察延迟是否明显上涨

docker exec -it redis redis-benchmark -a 123456 -c 200 -n 200000 -t get

2.6、补充对比提醒

之前的 Python 脚本受 GIL、网络开销、python 客户端限制，跑不出 10 万 QPS。

redis-benchmark：C 原生，测 Redis 理论极限； Python 脚本：适合模拟真实业务逻辑（带复杂 key、序列化、异常重试），用来验证业务并发场景。