文章目录一、Python 压测脚本1.1、redis_concurrent_test.py1.2、关键说明1.3、进阶优化建议可选1.4、协程版本补充高并发推荐二、redis内部提供的命令2.1、命令 redis-benchmark2.2、参数解释2.3、压测结果解读GET 纯读100 并发10 万请求核心指标2.4、简单分析2.5、建议继续测试复制直接运行1纯 SET 写压测最有参考价值2读写混合默认场景 SET/GET 等多条命令3调高并发到 200观察延迟是否明显上涨2.6、补充对比提醒一、Python 压测脚本使用threading多线程模拟并发读写打印详细耗时、成功率、QPS日志支持自定义并发数、请求总量、读写比例。依赖redis-pypipinstallredis1.1、redis_concurrent_test.pyimportthreadingimporttimeimportredisfromtypingimportDict# 配置区 REDIS_HOST127.0.0.1REDIS_PORT6379REDIS_PASSWORDNoneREDIS_DB0CONCURRENT_THREADS20# 并发线程数TOTAL_REQUESTS10000# 总请求次数TEST_KEY_PREFIXtest_concurrent:# 读写比例 True写, False读调整读写混合场景WRITE_RATIO0.3# # 全局统计stat:Dict{success:0,fail:0,total_cost:0.0,lock:threading.Lock()}defcreate_redis_client()-redis.Redis:创建redis连接每个线程独立client避免线程安全问题returnredis.Redis(hostREDIS_HOST,portREDIS_PORT,passwordREDIS_PASSWORD,dbREDIS_DB,decode_responsesTrue,socket_timeout5)defworker(thread_id:int,req_per_thread:int):单个线程工作函数rcreate_redis_client()thread_starttime.perf_counter()print(f[线程{thread_id:02d}] 启动计划执行{req_per_thread}次请求)foriinrange(req_per_thread):keyf{TEST_KEY_PREFIX}{thread_id}_{i}starttime.perf_counter()try:ifi/req_per_threadWRITE_RATIO:# 写操作 setr.setex(key,300,fvalue_{thread_id}_{i}_{time.time()})else:# 读操作 getr.get(key)costtime.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_costtime.perf_counter()-thread_startprint(f[线程{thread_id:02d}] 执行完成耗时{thread_cost:.3f}s)defmain():print(*60)print(fRedis单机并发压测启动)print(f地址:{REDIS_HOST}:{REDIS_PORT}| 并发线程:{CONCURRENT_THREADS})print(f总请求量:{TOTAL_REQUESTS}| 写比例:{WRITE_RATIO:.1%})print(*60)# 均分请求到各个线程base_reqTOTAL_REQUESTS//CONCURRENT_THREADS threads[]start_timetime.perf_counter()fortidinrange(CONCURRENT_THREADS):tthreading.Thread(targetworker,args(tid,base_req))threads.append(t)t.start()# 等待所有线程结束fortinthreads:t.join()total_timetime.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])*1000qpsstat[success]/total_timeprint(f平均延迟:{avg_latency_ms:.3f}ms)print(fQPS(成功):{qps:.2f}req/s)print(*60)if__name____main__:main()1.2、关键说明线程安全每个线程新建独立Redis实例禁止多线程共用同一个clientredis-py同步客户端非线程安全支持读写混合压测修改WRITE_RATIO1.0纯写压测0.0纯读压测日志输出每个线程启动/结束日志异常请求打印错误信息最终汇总总耗时、成功失败数、平均延迟、QPS适用于单机Redis如需集群压测只需修改连接初始化1.3、进阶优化建议可选如果想要更高压测性能使用gevent协程版本大量并发场景比thread高效开启redis pipeline批量请求使用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 get2.2、参数解释-a 123456Redis 认证密码 -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 p500.439ms一半请求低于该值 p950.815ms p991.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 set2读写混合默认场景 SET/GET 等多条命令docker exec -it redis redis-benchmark -a 123456 -c 100 -n 1000003调高并发到 200观察延迟是否明显上涨docker exec -it redis redis-benchmark -a 123456 -c 200 -n 200000 -t get2.6、补充对比提醒之前的 Python 脚本受 GIL、网络开销、python 客户端限制跑不出 10 万 QPS。redis-benchmarkC 原生测 Redis 理论极限 Python 脚本适合模拟真实业务逻辑带复杂 key、序列化、异常重试用来验证业务并发场景。