本文围绕「Redis 自身优化」「数据库架构扩展」「缓存数据库协同落地」三大核心方向结合压测中高频出现的性能瓶颈给出可直接落地的优化方案与完整代码实现所有方案均对应高并发场景下的典型瓶颈点可直接用于接口性能改造与压测优化验证。第一类Redis 自身性能深度优化实战配置代码双维度优化背景在 API 压测中Redis 层瓶颈通常表现为响应耗时毛刺、QPS 上不去、CPU 主线程阻塞、连接数打满、缓存命中率低。绝大多数问题并非 Redis 本身性能不足而是使用方式与配置不合理导致的性能浪费。本模块从客户端使用、数据结构优化、服务端配置、集群扩容四个维度完成 Redis 全链路性能调优。1. 连接池化与客户端参数调优对应瓶颈频繁创建销毁连接开销大、连接数耗尽导致请求排队、高并发下客户端超时。核心思路复用 TCP 连接控制最大连接数避免连接泄漏与频繁建连。Python 技术栈redis-py连接池完整实现importredisfromredis.connectionimportConnectionPool# 全局单例连接池避免每个请求新建连接池classRedisPool:_instanceNone_poolNonedef__new__(cls):ifcls._instanceisNone:cls._instancesuper().__new__(cls)# 连接池核心参数调优cls._poolConnectionPool(host127.0.0.1,port6379,passwordyour_password,db0,max_connections200,# 最大连接数与服务端 maxclients 匹配socket_timeout2,# socket 超时避免长时间阻塞socket_connect_timeout1,# 连接超时decode_responsesTrue,retry_on_timeoutTrue# 超时自动重试1次)returncls._instancedefget_client(self):returnredis.Redis(connection_poolself._pool)# 业务代码调用if__name____main__:redis_clientRedisPool().get_client()# 压测验证复用连接QPS 可提升 30%redis_client.set(user:1001,test_value,ex3600)print(redis_client.get(user:1001))2. 大Key拆解与慢查询治理对应瓶颈大Key读写阻塞主线程、内存占用不均、集群分片倾斜、慢查询拉低整体QPS。核心思路拆分大Hash/大List为小粒度Key避免单Key数据量过大通过慢查询日志定位低效命令。大Hash拆分实战代码defset_large_hash(user_id,data_dict,batch_size10): 将大Hash按字段拆分避免单Key过大 原方案hset(user:info, user_id, json.dumps(data_dict)) 优化后按业务字段分组拆分为多个小Hash redis_cliRedisPool().get_client()# 按字段拆分为基础信息、扩展信息两个小Keybase_fields[nickname,avatar,level]ext_fields[sign,tags,preference]base_data{k:vfork,vindata_dict.items()ifkinbase_fields}ext_data{k:vfork,vindata_dict.items()ifkinext_fields}# 管道批量执行减少网络IOpiperedis_cli.pipeline()ifbase_data:pipe.hset(fuser:base:{user_id},mappingbase_data)pipe.expire(fuser:base:{user_id},86400)ifext_data:pipe.hset(fuser:ext:{user_id},mappingext_data)pipe.expire(fuser:ext:{user_id},86400)pipe.execute()# 慢查询排查命令服务端执行# SLOWLOG GET 10 # 查看最近10条慢查询# CONFIG SET slowlog-log-slower-than 1000 # 慢查询阈值设为1ms3. 持久化与内存策略调优服务端配置对应瓶颈RDB/AOF 持久化引发周期性卡顿、内存淘汰策略不合理导致缓存命中率下降。核心配置redis.conf# 内存淘汰策略优先淘汰设置了过期时间的Key避免正常缓存被清理 maxmemory-policy volatile-lru # 内存上限建议设为物理内存的70%预留内存给fork与系统 maxmemory 8gb # 持久化优化高并发读场景降低持久化频率避免IO阻塞 # RDB 快照从默认3次触发调整为低频触发 save 900 1 save 300 10 # 关闭 AOF 实时刷盘改为每秒刷盘兼顾性能与数据安全 appendonly yes appendfsync everysec no-appendfsync-on-rewrite yes # 关闭透明大页避免内存分配延迟 # echo never /sys/kernel/mm/transparent_hugepage/enabled4. 集群分片水平扩容对应瓶颈单实例内存、QPS达到上限无法支撑更高并发。核心思路通过 Redis Cluster 实现数据分片水平扩展性能与容量。Python 客户端集群接入代码fromredis.clusterimportRedisCluster# 集群模式客户端自动路由Key到对应分片defget_cluster_client():startup_nodes[{host:192.168.1.10,port:6379},{host:192.168.1.11,port:6379},{host:192.168.1.12,port:6379}]returnRedisCluster(startup_nodesstartup_nodes,passwordyour_password,decode_responsesTrue,max_connections100)# 集群读写自动路由单实例瓶颈可通过扩容节点线性提升if__name____main__:clusterget_cluster_client()cluster.set(order:2001,paid,ex7200)print(cluster.get(order:2001))第二类数据库架构扩展实战从单库到高并发可扩展架构优化背景压测中 80% 的性能瓶颈最终落在数据库层单库单表在数据量增长、并发升高后会出现 IO 打满、CPU 飙升、锁冲突严重等问题。架构扩展的核心目标是分散压力、拆分负载、让数据库能力可水平扩展。本模块覆盖读写分离、水平分库分表、冷热数据分离三大核心扩展方案。1. 读写分离架构落地对应瓶颈读请求占比高90%单库读IO饱和QPS 被读能力限制。核心思路主库负责写多个从库负责读读能力随从库数量线性提升。Python SQLAlchemy 读写分离路由实现fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmakerimportrandom# 主库负责写操作MASTER_DBmysqlpymysql://user:passmaster-db:3306/shop?charsetutf8mb4# 从库列表负责读操作可水平扩容SLAVE_DBS[mysqlpymysql://user:passslave-db1:3306/shop?charsetutf8mb4,mysqlpymysql://user:passslave-db2:3306/shop?charsetutf8mb4]classDBRouter:def__init__(self):# 主库引擎self.master_enginecreate_engine(MASTER_DB,pool_size50,max_overflow100,pool_recycle3600)# 从库引擎列表self.slave_engines[create_engine(url,pool_size50,max_overflow100,pool_recycle3600)forurlinSLAVE_DBS]defget_master_session(self):获取写会话增删改、事务操作Sessionsessionmaker(bindself.master_engine)returnSession()defget_slave_session(self):获取读会话随机负载均衡到从库slave_enginerandom.choice(self.slave_engines)Sessionsessionmaker(bindslave_engine)returnSession()# 业务层使用示例if__name____main__:db_routerDBRouter()# 读请求走从库read_sessiondb_router.get_slave_session()resultread_session.execute(SELECT * FROM goods WHERE id 1001).fetchone()print(查询结果:,result)read_session.close()# 写请求走主库write_sessiondb_router.get_master_session()write_session.execute(UPDATE goods SET stock stock -1 WHERE id 1001)write_session.commit()write_session.close()2. 水平分库分表实战对应瓶颈单表数据量超千万索引效率下降深分页、全表扫描耗时剧增单库IO瓶颈。核心思路按业务字段如用户ID、订单ID水平拆分将数据分散到多张表/多个库中。分表路由逻辑 按用户ID分表示例classTableRouter:def__init__(self,table_base_name,shard_count8):self.table_basetable_base_name# 基础表名如 orderself.shard_countshard_count# 分表数量建议2的幂次defget_table_name(self,shard_key):根据分片键计算目标表名# 取模路由简单高效数据均匀分布shard_indexhash(shard_key)%self.shard_countreturnf{self.table_base}_{shard_index:02d}# 订单表分表示例按用户ID分8张表order_routerTableRouter(t_order,shard_count8)defquery_user_orders(user_id,page1,size20):分表查询自动路由到对应分表table_nameorder_router.get_table_name(user_id)sqlf SELECT order_id, amount, create_time FROM{table_name}WHERE user_id %s ORDER BY create_time DESC LIMIT %s OFFSET %s sessionDBRouter().get_slave_session()resultsession.execute(sql,(user_id,size,(page-1)*size)).fetchall()session.close()returnresult# 调用示例if__name____main__:ordersquery_user_orders(user_id10086)print(orders)企业级方案补充生产环境推荐接入 ShardingSphere-JDBC / ShardingSphere-Proxy通过配置化实现分库分表、读写分离无需手动编写路由逻辑核心配置示例# ShardingSphere 分表配置rules:-!SHARDINGtables:t_order:actualDataNodes:ds0.t_order_${0..7}tableStrategy:standard:shardingColumn:user_idshardingAlgorithmName:order_inlineshardingAlgorithms:order_inline:type:INLINEprops:algorithm-expression:t_order_${user_id % 8}3. 冷热数据分离与归档对应瓶颈单表历史数据过多查询扫描范围大索引效率低占用大量IO与内存。核心思路将低频访问的历史数据归档到历史库主表只保留近3-6个月热数据。数据归档与查询路由代码defget_order_table_by_time(create_time):按时间路由近3个月走热库主表更早走冷库归档表fromdatetimeimportdatetime,timedelta hot_deadlinedatetime.now()-timedelta(days90)ifcreate_timehot_deadline:returnt_order,hot_db# 热库主表else:returnt_order_history,cold_db# 冷库归档表defarchive_history_data(before_days90):定时归档任务迁移历史数据到归档表sql_movef INSERT INTO t_order_history SELECT * FROM t_order WHERE create_time DATE_SUB(NOW(), INTERVAL{before_days}DAY) sql_deletef DELETE FROM t_order WHERE create_time DATE_SUB(NOW(), INTERVAL{before_days}DAY) sessionDBRouter().get_master_session()session.execute(sql_move)session.execute(sql_delete)session.commit()session.close()第三类缓存数据库协同高并发架构解决经典性能与一致性问题优化背景单独优化 Redis 或数据库无法彻底解决高并发问题两者协同不当会引发缓存击穿、雪崩、数据不一致等次生问题。本类聚焦两者配合的标准架构在保证性能的同时兼顾数据正确性是压测后落地优化的核心环节。1. 标准 Cache Aside 模式读写实现对应瓶颈缓存与数据库更新顺序错误导致数据不一致缓存更新逻辑冗余增加接口耗时。核心规则读时先查缓存缓存没有再查数据库并回写缓存更新时先更新数据库再删除缓存。完整业务代码实现classGoodsService:def__init__(self):self.redisRedisPool().get_client()self.db_routerDBRouter()self.cache_key_prefixgoods:info:self.cache_expire3600# 缓存过期时间1小时defget_goods_info(self,goods_id):读接口Cache Aside 标准流程cache_keyf{self.cache_key_prefix}{goods_id}# 1. 先查缓存cache_dataself.redis.get(cache_key)ifcache_data:returncache_data# 缓存命中直接返回# 2. 缓存未命中查数据库db_sessionself.db_router.get_slave_session()sqlSELECT id, name, price, stock FROM goods WHERE id %sgoodsdb_session.execute(sql,(goods_id,)).fetchone()db_session.close()ifnotgoods:# 空值缓存防止缓存穿透过期时间缩短self.redis.setex(cache_key,60,)returnNone# 3. 回写缓存设置过期时间goods_strf{goods.id}|{goods.name}|{goods.price}|{goods.stock}self.redis.setex(cache_key,self.cache_expire,goods_str)returngoods_strdefupdate_goods_price(self,goods_id,new_price):更新接口先更数据库再删缓存# 1. 更新数据库db_sessionself.db_router.get_master_session()sqlUPDATE goods SET price %s WHERE id %sdb_session.execute(sql,(new_price,goods_id))db_session.commit()db_session.close()# 2. 删除缓存下次读取自动回写最新数据cache_keyf{self.cache_key_prefix}{goods_id}self.redis.delete(cache_key)returnTrue2. 分布式锁解决缓存击穿问题对应瓶颈热点Key失效瞬间大量并发请求同时穿透到数据库导致数据库瞬间压力飙升缓存击穿。核心思路缓存重建时加分布式锁同一时间只有一个请求去查库回写缓存其余请求等待重试。Redis 分布式锁防击穿实现importtimeimportuuidclassCacheService:def__init__(self):self.redisRedisPool().get_client()self.lock_prefixlock:cache:self.lock_timeout3# 锁超时时间避免死锁defget_data_with_lock(self,key,query_db_func,expire3600):带互斥锁的缓存读取防止热点Key击穿# 1. 正常查询缓存dataself.redis.get(key)ifdataisnotNoneanddata!:returndata# 2. 缓存未命中尝试加锁lock_keyf{self.lock_prefix}{key}request_idstr(uuid.uuid4())lock_successself.redis.set(lock_key,request_id,exself.lock_timeout,nxTrue)iflock_success:try:# 3. 拿到锁查询数据库并回写缓存dataquery_db_func()ifdata:self.redis.setex(key,expire,data)else:self.redis.setex(key,60,)# 空值缓存returndatafinally:# 4. 释放锁只能释放自己加的锁ifself.redis.get(lock_key)request_id:self.redis.delete(lock_key)else:# 5. 没拿到锁休眠后重试time.sleep(0.1)returnself.get_data_with_lock(key,query_db_func,expire)# 业务调用示例if__name____main__:cache_svcCacheService()defquery_goods_from_db():# 模拟数据库查询returngoods_info_1001# 高并发下只有一个请求会查库其余等待缓存resultcache_svc.get_data_with_lock(goods:1001,query_goods_from_db)print(result)3. 最终一致性保障异步更新缓存对应瓶颈高频写场景下每次删缓存仍有短暂不一致窗口高并发写删缓存频繁性能损耗大。核心思路数据库更新后通过消息队列异步更新缓存保证最终一致降低主接口耗时。MQ 异步更新缓存伪代码# 1. 更新接口只更数据库发送更新消息defupdate_goods(goods_id,new_data):db_sessionDBRouter().get_master_session()# 更新数据库db_session.execute(UPDATE goods SET ... WHERE id %s,(goods_id,))db_session.commit()db_session.close()# 发送消息到MQ异步更新缓存send_mq(topiccache_update_topic,body{type:goods,id:goods_id})returnTrue# 2. 消费者监听消息异步更新缓存defcache_update_consumer(msg):datamsg.body goods_iddata[id]# 查询最新数据db_sessionDBRouter().get_slave_session()goodsdb_session.execute(SELECT * FROM goods WHERE id %s,(goods_id,)).fetchone()db_session.close()# 更新缓存redis_cliRedisPool().get_client()redis_cli.setex(fgoods:info:{goods_id},3600,str(goods))优化效果验证标准所有优化完成后需通过压测复现验证核心对比指标Redis 层命中率 ≥ 95%平均响应耗时 1ms无慢查询连接数稳定不溢出数据库层CPU 使用率 ≤ 70%IO 使用率 ≤ 60%慢SQL数量降为0锁等待大幅减少接口层QPS 提升 2~10 倍P99 响应时间下降 50%错误率 ≤ 0.1%
高并发架构优化实战：Redis 调优、数据库扩展与协同架构三大核心模块
发布时间：2026/6/30 4:47:10
本文围绕「Redis 自身优化」「数据库架构扩展」「缓存数据库协同落地」三大核心方向结合压测中高频出现的性能瓶颈给出可直接落地的优化方案与完整代码实现所有方案均对应高并发场景下的典型瓶颈点可直接用于接口性能改造与压测优化验证。第一类Redis 自身性能深度优化实战配置代码双维度优化背景在 API 压测中Redis 层瓶颈通常表现为响应耗时毛刺、QPS 上不去、CPU 主线程阻塞、连接数打满、缓存命中率低。绝大多数问题并非 Redis 本身性能不足而是使用方式与配置不合理导致的性能浪费。本模块从客户端使用、数据结构优化、服务端配置、集群扩容四个维度完成 Redis 全链路性能调优。1. 连接池化与客户端参数调优对应瓶颈频繁创建销毁连接开销大、连接数耗尽导致请求排队、高并发下客户端超时。核心思路复用 TCP 连接控制最大连接数避免连接泄漏与频繁建连。Python 技术栈redis-py连接池完整实现importredisfromredis.connectionimportConnectionPool# 全局单例连接池避免每个请求新建连接池classRedisPool:_instanceNone_poolNonedef__new__(cls):ifcls._instanceisNone:cls._instancesuper().__new__(cls)# 连接池核心参数调优cls._poolConnectionPool(host127.0.0.1,port6379,passwordyour_password,db0,max_connections200,# 最大连接数与服务端 maxclients 匹配socket_timeout2,# socket 超时避免长时间阻塞socket_connect_timeout1,# 连接超时decode_responsesTrue,retry_on_timeoutTrue# 超时自动重试1次)returncls._instancedefget_client(self):returnredis.Redis(connection_poolself._pool)# 业务代码调用if__name____main__:redis_clientRedisPool().get_client()# 压测验证复用连接QPS 可提升 30%redis_client.set(user:1001,test_value,ex3600)print(redis_client.get(user:1001))2. 大Key拆解与慢查询治理对应瓶颈大Key读写阻塞主线程、内存占用不均、集群分片倾斜、慢查询拉低整体QPS。核心思路拆分大Hash/大List为小粒度Key避免单Key数据量过大通过慢查询日志定位低效命令。大Hash拆分实战代码defset_large_hash(user_id,data_dict,batch_size10): 将大Hash按字段拆分避免单Key过大 原方案hset(user:info, user_id, json.dumps(data_dict)) 优化后按业务字段分组拆分为多个小Hash redis_cliRedisPool().get_client()# 按字段拆分为基础信息、扩展信息两个小Keybase_fields[nickname,avatar,level]ext_fields[sign,tags,preference]base_data{k:vfork,vindata_dict.items()ifkinbase_fields}ext_data{k:vfork,vindata_dict.items()ifkinext_fields}# 管道批量执行减少网络IOpiperedis_cli.pipeline()ifbase_data:pipe.hset(fuser:base:{user_id},mappingbase_data)pipe.expire(fuser:base:{user_id},86400)ifext_data:pipe.hset(fuser:ext:{user_id},mappingext_data)pipe.expire(fuser:ext:{user_id},86400)pipe.execute()# 慢查询排查命令服务端执行# SLOWLOG GET 10 # 查看最近10条慢查询# CONFIG SET slowlog-log-slower-than 1000 # 慢查询阈值设为1ms3. 持久化与内存策略调优服务端配置对应瓶颈RDB/AOF 持久化引发周期性卡顿、内存淘汰策略不合理导致缓存命中率下降。核心配置redis.conf# 内存淘汰策略优先淘汰设置了过期时间的Key避免正常缓存被清理 maxmemory-policy volatile-lru # 内存上限建议设为物理内存的70%预留内存给fork与系统 maxmemory 8gb # 持久化优化高并发读场景降低持久化频率避免IO阻塞 # RDB 快照从默认3次触发调整为低频触发 save 900 1 save 300 10 # 关闭 AOF 实时刷盘改为每秒刷盘兼顾性能与数据安全 appendonly yes appendfsync everysec no-appendfsync-on-rewrite yes # 关闭透明大页避免内存分配延迟 # echo never /sys/kernel/mm/transparent_hugepage/enabled4. 集群分片水平扩容对应瓶颈单实例内存、QPS达到上限无法支撑更高并发。核心思路通过 Redis Cluster 实现数据分片水平扩展性能与容量。Python 客户端集群接入代码fromredis.clusterimportRedisCluster# 集群模式客户端自动路由Key到对应分片defget_cluster_client():startup_nodes[{host:192.168.1.10,port:6379},{host:192.168.1.11,port:6379},{host:192.168.1.12,port:6379}]returnRedisCluster(startup_nodesstartup_nodes,passwordyour_password,decode_responsesTrue,max_connections100)# 集群读写自动路由单实例瓶颈可通过扩容节点线性提升if__name____main__:clusterget_cluster_client()cluster.set(order:2001,paid,ex7200)print(cluster.get(order:2001))第二类数据库架构扩展实战从单库到高并发可扩展架构优化背景压测中 80% 的性能瓶颈最终落在数据库层单库单表在数据量增长、并发升高后会出现 IO 打满、CPU 飙升、锁冲突严重等问题。架构扩展的核心目标是分散压力、拆分负载、让数据库能力可水平扩展。本模块覆盖读写分离、水平分库分表、冷热数据分离三大核心扩展方案。1. 读写分离架构落地对应瓶颈读请求占比高90%单库读IO饱和QPS 被读能力限制。核心思路主库负责写多个从库负责读读能力随从库数量线性提升。Python SQLAlchemy 读写分离路由实现fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmakerimportrandom# 主库负责写操作MASTER_DBmysqlpymysql://user:passmaster-db:3306/shop?charsetutf8mb4# 从库列表负责读操作可水平扩容SLAVE_DBS[mysqlpymysql://user:passslave-db1:3306/shop?charsetutf8mb4,mysqlpymysql://user:passslave-db2:3306/shop?charsetutf8mb4]classDBRouter:def__init__(self):# 主库引擎self.master_enginecreate_engine(MASTER_DB,pool_size50,max_overflow100,pool_recycle3600)# 从库引擎列表self.slave_engines[create_engine(url,pool_size50,max_overflow100,pool_recycle3600)forurlinSLAVE_DBS]defget_master_session(self):获取写会话增删改、事务操作Sessionsessionmaker(bindself.master_engine)returnSession()defget_slave_session(self):获取读会话随机负载均衡到从库slave_enginerandom.choice(self.slave_engines)Sessionsessionmaker(bindslave_engine)returnSession()# 业务层使用示例if__name____main__:db_routerDBRouter()# 读请求走从库read_sessiondb_router.get_slave_session()resultread_session.execute(SELECT * FROM goods WHERE id 1001).fetchone()print(查询结果:,result)read_session.close()# 写请求走主库write_sessiondb_router.get_master_session()write_session.execute(UPDATE goods SET stock stock -1 WHERE id 1001)write_session.commit()write_session.close()2. 水平分库分表实战对应瓶颈单表数据量超千万索引效率下降深分页、全表扫描耗时剧增单库IO瓶颈。核心思路按业务字段如用户ID、订单ID水平拆分将数据分散到多张表/多个库中。分表路由逻辑 按用户ID分表示例classTableRouter:def__init__(self,table_base_name,shard_count8):self.table_basetable_base_name# 基础表名如 orderself.shard_countshard_count# 分表数量建议2的幂次defget_table_name(self,shard_key):根据分片键计算目标表名# 取模路由简单高效数据均匀分布shard_indexhash(shard_key)%self.shard_countreturnf{self.table_base}_{shard_index:02d}# 订单表分表示例按用户ID分8张表order_routerTableRouter(t_order,shard_count8)defquery_user_orders(user_id,page1,size20):分表查询自动路由到对应分表table_nameorder_router.get_table_name(user_id)sqlf SELECT order_id, amount, create_time FROM{table_name}WHERE user_id %s ORDER BY create_time DESC LIMIT %s OFFSET %s sessionDBRouter().get_slave_session()resultsession.execute(sql,(user_id,size,(page-1)*size)).fetchall()session.close()returnresult# 调用示例if__name____main__:ordersquery_user_orders(user_id10086)print(orders)企业级方案补充生产环境推荐接入 ShardingSphere-JDBC / ShardingSphere-Proxy通过配置化实现分库分表、读写分离无需手动编写路由逻辑核心配置示例# ShardingSphere 分表配置rules:-!SHARDINGtables:t_order:actualDataNodes:ds0.t_order_${0..7}tableStrategy:standard:shardingColumn:user_idshardingAlgorithmName:order_inlineshardingAlgorithms:order_inline:type:INLINEprops:algorithm-expression:t_order_${user_id % 8}3. 冷热数据分离与归档对应瓶颈单表历史数据过多查询扫描范围大索引效率低占用大量IO与内存。核心思路将低频访问的历史数据归档到历史库主表只保留近3-6个月热数据。数据归档与查询路由代码defget_order_table_by_time(create_time):按时间路由近3个月走热库主表更早走冷库归档表fromdatetimeimportdatetime,timedelta hot_deadlinedatetime.now()-timedelta(days90)ifcreate_timehot_deadline:returnt_order,hot_db# 热库主表else:returnt_order_history,cold_db# 冷库归档表defarchive_history_data(before_days90):定时归档任务迁移历史数据到归档表sql_movef INSERT INTO t_order_history SELECT * FROM t_order WHERE create_time DATE_SUB(NOW(), INTERVAL{before_days}DAY) sql_deletef DELETE FROM t_order WHERE create_time DATE_SUB(NOW(), INTERVAL{before_days}DAY) sessionDBRouter().get_master_session()session.execute(sql_move)session.execute(sql_delete)session.commit()session.close()第三类缓存数据库协同高并发架构解决经典性能与一致性问题优化背景单独优化 Redis 或数据库无法彻底解决高并发问题两者协同不当会引发缓存击穿、雪崩、数据不一致等次生问题。本类聚焦两者配合的标准架构在保证性能的同时兼顾数据正确性是压测后落地优化的核心环节。1. 标准 Cache Aside 模式读写实现对应瓶颈缓存与数据库更新顺序错误导致数据不一致缓存更新逻辑冗余增加接口耗时。核心规则读时先查缓存缓存没有再查数据库并回写缓存更新时先更新数据库再删除缓存。完整业务代码实现classGoodsService:def__init__(self):self.redisRedisPool().get_client()self.db_routerDBRouter()self.cache_key_prefixgoods:info:self.cache_expire3600# 缓存过期时间1小时defget_goods_info(self,goods_id):读接口Cache Aside 标准流程cache_keyf{self.cache_key_prefix}{goods_id}# 1. 先查缓存cache_dataself.redis.get(cache_key)ifcache_data:returncache_data# 缓存命中直接返回# 2. 缓存未命中查数据库db_sessionself.db_router.get_slave_session()sqlSELECT id, name, price, stock FROM goods WHERE id %sgoodsdb_session.execute(sql,(goods_id,)).fetchone()db_session.close()ifnotgoods:# 空值缓存防止缓存穿透过期时间缩短self.redis.setex(cache_key,60,)returnNone# 3. 回写缓存设置过期时间goods_strf{goods.id}|{goods.name}|{goods.price}|{goods.stock}self.redis.setex(cache_key,self.cache_expire,goods_str)returngoods_strdefupdate_goods_price(self,goods_id,new_price):更新接口先更数据库再删缓存# 1. 更新数据库db_sessionself.db_router.get_master_session()sqlUPDATE goods SET price %s WHERE id %sdb_session.execute(sql,(new_price,goods_id))db_session.commit()db_session.close()# 2. 删除缓存下次读取自动回写最新数据cache_keyf{self.cache_key_prefix}{goods_id}self.redis.delete(cache_key)returnTrue2. 分布式锁解决缓存击穿问题对应瓶颈热点Key失效瞬间大量并发请求同时穿透到数据库导致数据库瞬间压力飙升缓存击穿。核心思路缓存重建时加分布式锁同一时间只有一个请求去查库回写缓存其余请求等待重试。Redis 分布式锁防击穿实现importtimeimportuuidclassCacheService:def__init__(self):self.redisRedisPool().get_client()self.lock_prefixlock:cache:self.lock_timeout3# 锁超时时间避免死锁defget_data_with_lock(self,key,query_db_func,expire3600):带互斥锁的缓存读取防止热点Key击穿# 1. 正常查询缓存dataself.redis.get(key)ifdataisnotNoneanddata!:returndata# 2. 缓存未命中尝试加锁lock_keyf{self.lock_prefix}{key}request_idstr(uuid.uuid4())lock_successself.redis.set(lock_key,request_id,exself.lock_timeout,nxTrue)iflock_success:try:# 3. 拿到锁查询数据库并回写缓存dataquery_db_func()ifdata:self.redis.setex(key,expire,data)else:self.redis.setex(key,60,)# 空值缓存returndatafinally:# 4. 释放锁只能释放自己加的锁ifself.redis.get(lock_key)request_id:self.redis.delete(lock_key)else:# 5. 没拿到锁休眠后重试time.sleep(0.1)returnself.get_data_with_lock(key,query_db_func,expire)# 业务调用示例if__name____main__:cache_svcCacheService()defquery_goods_from_db():# 模拟数据库查询returngoods_info_1001# 高并发下只有一个请求会查库其余等待缓存resultcache_svc.get_data_with_lock(goods:1001,query_goods_from_db)print(result)3. 最终一致性保障异步更新缓存对应瓶颈高频写场景下每次删缓存仍有短暂不一致窗口高并发写删缓存频繁性能损耗大。核心思路数据库更新后通过消息队列异步更新缓存保证最终一致降低主接口耗时。MQ 异步更新缓存伪代码# 1. 更新接口只更数据库发送更新消息defupdate_goods(goods_id,new_data):db_sessionDBRouter().get_master_session()# 更新数据库db_session.execute(UPDATE goods SET ... WHERE id %s,(goods_id,))db_session.commit()db_session.close()# 发送消息到MQ异步更新缓存send_mq(topiccache_update_topic,body{type:goods,id:goods_id})returnTrue# 2. 消费者监听消息异步更新缓存defcache_update_consumer(msg):datamsg.body goods_iddata[id]# 查询最新数据db_sessionDBRouter().get_slave_session()goodsdb_session.execute(SELECT * FROM goods WHERE id %s,(goods_id,)).fetchone()db_session.close()# 更新缓存redis_cliRedisPool().get_client()redis_cli.setex(fgoods:info:{goods_id},3600,str(goods))优化效果验证标准所有优化完成后需通过压测复现验证核心对比指标Redis 层命中率 ≥ 95%平均响应耗时 1ms无慢查询连接数稳定不溢出数据库层CPU 使用率 ≤ 70%IO 使用率 ≤ 60%慢SQL数量降为0锁等待大幅减少接口层QPS 提升 2~10 倍P99 响应时间下降 50%错误率 ≤ 0.1%
![华为OD机试2025C卷-字符串变换最小次数[100分]（ Java _ Python3 _ C++ _ C语言 _ JsNode _ Go）实现100%通过率](http://pic.xiahunao.cn/yaotu/华为OD机试2025C卷-字符串变换最小次数[100分]（ Java _ Python3 _ C++ _ C语言 _ JsNode _ Go）实现100%通过率)
![华为OD机试2025C卷-内存资源分配[100分]（ Java _ Python3 _ C++ _ C语言 _ JsNode _ Go）实现100%通过率](http://pic.xiahunao.cn/yaotu/华为OD机试2025C卷-内存资源分配[100分]（ Java _ Python3 _ C++ _ C语言 _ JsNode _ Go）实现100%通过率)
