如何利用asyncio.Lock避免Redis数据加载操作重复执行

问题分析 #

你的原代码存在竞态条件：当多个协程同时通过锁外的if self.lock.locked()检查后，会依次进入锁内执行加载逻辑，导致重复加载。虽然在锁内二次检查能解决问题，但你希望有更清晰的实现方式。

标准优化方案：清晰化的双重检查锁定 #

双重检查锁定本身是这类场景的标准解法，我们可以通过调整代码结构让逻辑更直观，避免冗余判断：

import asyncio

class MyService:
    def __init__(self):
        self.lock = asyncio.Lock()
        self.latest_timestamp = None

    async def _check_latest_timestamp_from_db(self):
        # 从Redis获取最新时间戳的逻辑
        # 示例：return await redis.get("data_update_timestamp")
        return "2024-05-20T12:00:00"

    async def ensure_update_loaded(self):
        # 第一步：无锁快速检查，无更新直接返回
        latest_timestamp_from_db = await self._check_latest_timestamp_from_db()
        if self.latest_timestamp == latest_timestamp_from_db:
            return
        
        # 第二步：加锁后再次确认，避免竞态
        async with self.lock:
            # 加锁后必须二次检查，防止其他协程已完成更新
            if self.latest_timestamp == latest_timestamp_from_db:
                return
            # 执行数据加载逻辑
            print("开始加载最新数据...")
            # 模拟加载操作
            await asyncio.sleep(1)
            self.latest_timestamp = latest_timestamp_from_db
            print("数据加载完成")

方案优势 #

• 逻辑分层明确：先做无锁快速判断，过滤掉绝大多数无需加载的请求
• 锁内二次检查是必须环节：第一次检查到加锁的间隙，可能已有其他协程完成了更新
• 代码结构简洁，去掉了多余的lock.locked()判断，async with会自动处理锁的获取，后续协程会等待锁释放后再进入检查

可选：让等待协程直接返回（不等待加载完成） #

如果业务允许后续协程直接使用旧数据、无需等待加载完成，可以在加锁前增加锁状态检查：

async def ensure_update_loaded(self):
    latest_timestamp_from_db = await self._check_latest_timestamp_from_db()
    if self.latest_timestamp == latest_timestamp_from_db:
        return
    
    # 已有协程在加载时直接返回，不等待
    if self.lock.locked():
        return
    
    async with self.lock:
        if self.latest_timestamp == latest_timestamp_from_db:
            return
        # 执行加载逻辑
        await self._load_data_from_db()
        self.latest_timestamp = latest_timestamp_from_db

这种写法保留了双重检查逻辑，同时避免了不必要的等待，但仅适用于允许临时使用旧数据的场景。

其他可选方案：用事件标记加载状态 #

如果觉得锁的写法不够直观，可使用asyncio.Event标记加载状态，本质逻辑和双重检查一致：

class MyService:
    def __init__(self):
        self.loading_event = asyncio.Event()
        self.loading_event.set()  # 初始状态为未加载
        self.latest_timestamp = None

    async def ensure_update_loaded(self):
        latest_timestamp_from_db = await self._check_latest_timestamp_from_db()
        if self.latest_timestamp == latest_timestamp_from_db:
            return
        
        # 正在加载时等待完成后直接返回
        if not self.loading_event.is_set():
            await self.loading_event.wait()
            return
        
        self.loading_event.clear()
        try:
            # 二次检查
            if self.latest_timestamp == latest_timestamp_from_db:
                return
            await self._load_data_from_db()
            self.latest_timestamp = latest_timestamp_from_db
        finally:
            self.loading_event.set()

这个方案用事件标记加载状态，逻辑同样清晰，但对于简单场景，锁的写法更轻量。

内容的提问来源于stack exchange，提问作者Artem Ilin