Redis热点key问题真是让人头大，尤其在大促期间，几个热门商品的缓存key被疯狂访问，分分钟就能把系统搞趴下。不过别担心，今天咱聊聊几个实用的热点key处理方案，能帮你把访问延迟降到最低。

本地缓存大法

在Redis前面加一层本地缓存，能有效减少网络开销：

from cachetools import TTLCache
import redis

class LocalCache:
    def __init__(self):
        self.local_cache = TTLCache(maxsize=1000, ttl=5)  # 5秒过期
        self.redis_client = redis.Redis()
    
    def get_product_info(self, product_id: str):
        # 先查本地缓存
        if product_id in self.local_cache:
            return self.local_cache[product_id]
        
        # 本地没有查Redis
        value = self.redis_client.get(f"product:{product_id}")
        if value:
            self.local_cache[product_id] = value
        return value

温馨提示：本地缓存过期时间别设太长，不然数据不一致问题会让你很头疼。

分片打散技术

把一个热key分成多个子key，分散访问压力：

class ShardingCache:
    def __init__(self, shard_num=10):
        self.shard_num = shard_num
        self.redis_client = redis.Redis()
    
    def get_hot_key(self, key: str):
        # 随机选择一个分片
        shard = random.randint(0, self.shard_num - 1)
        shard_key = f"{key}:{shard}"
        
        return self.redis_client.get(shard_key)
    
    def set_hot_key(self, key: str, value: str):
        # 写入所有分片
        for i in range(self.shard_num):
            shard_key = f"{key}:{i}"
            self.redis_client.set(shard_key, value)

说实话，这招在我们公司双11时立了大功，原来动不动几百毫秒的接口直接降到了10毫秒以内。

异步读取优化

用异步方式读取热点数据，能大幅提升并发性能：

import asyncio
import aioredis

class AsyncHotKeyHandler:
    def __init__(self):
        self.redis = aioredis.from_url('redis://localhost')
    
    async def get_hot_data(self, keys: list):
        # 批量异步读取
        tasks = [self.redis.get(key) for key in keys]
        results = await asyncio.gather(*tasks)
        return dict(zip(keys, results))

温馨提示：异步读取时要注意控制并发量，不要一次性发太多请求。

延迟加载策略

有时候可以先返回旧数据，后台更新：

from threading import Thread
import time

class LazyLoader:
    def __init__(self):
        self.cache = {}
        self.redis_client = redis.Redis()
    
    def get_data(self, key: str):
        cached = self.cache.get(key)
        if cached:
            # 异步更新缓存
            Thread(target=self._async_refresh, args=(key,)).start()
            return cached
        
        # 缓存不存在时直接查Redis
        return self._load_from_redis(key)
    
    def _async_refresh(self, key: str):
        new_value = self._load_from_redis(key)
        self.cache[key] = new_value

记得去年我遇到个有意思的case，一个商品详情页的热key每秒几万次访问，直接把Redis干懵了。上了这套方案后，Redis的压力直接降了90%。

要是你想玩得更花，还可以用布隆过滤器预加载热key，提前把可能的热点数据写入本地缓存。不过话说回来，系统的复杂度也上来了，得好好权衡。

写代码也要动动脑子，比如定时任务更新热点数据时，可以错开整点，避免跟业务高峰撞到一起。我看过好多系统半夜12点集中更新缓存，结果一更新全站都卡。

有个小技巧，更新热点数据时最好用SETNX，不要直接SET，这样能避免多个实例同时更新导致的问题。

还有个坑，不少人喜欢用incr命令更新热点计数器，这种情况最好用批量操作，积累一批后一次性更新，不然Redis吃不消。

实际应用中啊，这些方案都不是单独用的，得根据业务特点组合使用。我们生产环境就是本地缓存+分片+异步读取一起上，效果相当不错。