如何给包含数据库读写、API调用的复合操作加锁避免并发冲突？

针对你的场景，分两种部署情况给出可落地的实现方案： #

方案1：单进程部署（优先选择，性能最优） #

你使用的是async异步写法，单进程部署下可以通过asyncio.Lock实现按经纬度维度的细粒度内存锁，既避免重复调用外部API，也不会影响无关经纬度的请求性能。

实现逻辑 #

• 维护一个全局字典存储每个经纬度对应的独立锁
• 请求进来先做一次无锁缓存查询，已命中直接返回
• 未命中时获取对应经纬度的锁，拿到锁后必须二次查询缓存（避免等待锁的过程中已经有其他请求完成了缓存写入）
• 确认无缓存再调用外部API，写入缓存后释放锁
• 增加引用计数清理闲置锁，避免内存泄漏

示例代码 #

import asyncio
from typing import Dict
from sqlalchemy.exc import IntegrityError

# 全局锁存储：key为经纬度拼接字符串，value为对应锁实例
poly_locks: Dict[str, asyncio.Lock] = {}
# 锁引用计数，用于清理无请求的闲置锁
lock_ref_count: Dict[str, int] = {}

async def _get_poly_id(self, longitude: str, latitude: str):
    # 无锁快速查询已缓存的结果
    poly_id = self._polygon_manager.get_poly_id(longitude=longitude, latitude=latitude)
    if poly_id is not None:
        return poly_id

    lock_key = f"{longitude}:{latitude}"
    # 不存在对应锁则新建
    if lock_key not in poly_locks:
        poly_locks[lock_key] = asyncio.Lock()
        lock_ref_count[lock_key] = 0
    lock = poly_locks[lock_key]
    lock_ref_count[lock_key] += 1

    try:
        async with lock:
            # 拿到锁后二次查询，避免等待期间缓存已被其他请求写入
            poly_id = self._polygon_manager.get_poly_id(longitude=longitude, latitude=latitude)
            if poly_id is not None:
                return poly_id
            # 确认无缓存再调用外部API
            geo_json = self._point2geojson(longitude=longitude, latitude=latitude)
            poly_id = await self._agro_api_helper.create_polygon(geo_json)
            self._polygon_manager.save_poly_id(longitude=longitude, latitude=latitude, poly_id=poly_id)
            return poly_id
    except IntegrityError:
        # 兜底逻辑：极端锁失效场景下冲突，直接读取已写入的缓存返回
        return self._polygon_manager.get_poly_id(longitude=longitude, latitude=latitude)
    finally:
        # 引用计数归零后清理锁，避免内存泄漏
        lock_ref_count[lock_key] -= 1
        if lock_ref_count[lock_key] == 0:
            del poly_locks[lock_key]
            del lock_ref_count[lock_key]

方案2：多进程/多实例部署 #

如果服务是多实例部署，内存锁不生效，可以通过数据库级锁实现：

1. 开启SQLAlchemy事务的可重复读隔离级别
2. 缓存未命中时开启事务，尝试对目标经纬度记录加排他锁，不存在则先插入状态为「待填充」的占位记录
3. 调用外部API成功后更新占位记录的poly_id字段
4. 其他请求拿到锁后如果发现记录已被填充，直接返回结果即可

需要注意SQLite本身的写锁是库级的，高并发场景下性能表现一般，如果业务并发量较高，更建议把缓存层换成Redis加分布式锁实现。

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