Redis缓存穿透,缓存击穿,缓存雪崩

class RDS():
    def __init__():
        # 注:__init__逻辑里面并不能使用await,该行代码无法正常运行,这里只是简要说明初始化连接池
        self.conn_pool = await aioredis.create_pool()
        
    async def execute(
            self, command: str, *args: Any, **kwargs: Any
    ) -> Optional[Any]:
        try:
            async with self.conn_pool.get() as conn:
                return await conn.execute(command, *args, **kwargs)
        except Exception as e:
        	pass # 抛错自定义,与演示无关,先pass
        return None
        
    @asynccontextmanager
    async def lock(self, name: str, timeout: int = 1 * 60) -> bool:
    	# 这里锁粒度是天,可以根据自己业务定义
        today_string: str = datetime.datetime.now().strftime("%Y-%m-%d") 
        key: str = f"project:lock:{name}:{today_string}"
        try:
            lock = await self.execute('set', key, '1', 'ex', timeout, 'nx')
            if lock == 1:
                lock = True
            else:
                lock = False
            yield lock
        finally:
            await self.execute('del', key)

rds = RDS()

async def api_demo():
    while:
        async with rds.lock as lock:
            if lock:
                pass # 获得锁的逻辑
            await async.sleep(0.01)

# 清理现场逻辑最好还是用计数器来保证所有调用调用结束后才进行清理现场,不过代码量会变多,所以这里简单的使用延迟清理逻辑来代替.
# 同时为了逻辑清晰,并没有全部使用asyncio.Future的功能
import asyncio
from functools import partial


class Share():
    def __init__(self, delay_clean_time=1):
        self._future_dict = dict()
        self._result_dict = dict()
        self._error_dict = dict()
        self._loop = asyncio.get_event_loop()
        self._delay_clean_time = delay_clean_time  # 代表执行完毕后多少秒清理现场

    async def clean(self, key):
        await asyncio.sleep(self._delay_clean_time)
        # 清理现场
        if key in self._future_dict:
            del self._future_dict[key]
        if key in self._result_dict:
            del self._result_dict[key]
        if key in self._error_dict:
            del self._error_dict[key]
    
    async def do(self, key, fn):
        if key not in self._future_dict:
            # 第一个执行的协程,进入执行逻辑
            self._future_dict[key] = asyncio.Future()  # 设置一个Future,其他协程可以等待这个Future直到他被执行set_result(True)
            try:
                self._result_dict[key] = await fn()  # 执行函数(如果有同步和异步函数,需要做判断,这里只示范异步)
            except Exception as e:
                self._error_dict[key] = e   # 记录异常
            finally:
                self._future_dict[key].set_result(True)  # 代表执行完毕
                
                # 执行延迟清理现场逻辑
                asyncio.ensure_future(self.clean(key))
        else:
            # 非第一个执行的协程进入等待逻辑
            await self._future_dict[key]
            
        # 有异常则抛异常
        if key in self._error_dict:
            raise self._error_dict[key]
        # 返回数据
        return self._result_dict[key]

rds = RDS()
share =  Share()
share_key = 'so1n'

async def get_data_demo():
    while:
        async with rds.lock as lock:
            if lock:
                pass # 获得锁的逻辑
            await async.sleep(0.01)
            
async def api_demo():
    return share.do(share_key, get_data_demo)