Python脚本如何加锁控制并发缓存写入

wen python案例 2026-06-26 32

本文目录导读：

使用 threading.Lock（线程安全）
使用 multiprocessing.Lock（进程安全）
使用 asyncio.Lock（异步安全）
使用 Redis 实现分布式锁（跨进程/跨机器）
使用 cachetools 库（推荐）
最佳实践建议

在Python中控制并发缓存写入,有几种常用方法：

使用 threading.Lock（线程安全）

import threading
import time
from functools import lru_cache
class ThreadSafeCache:
    def __init__(self):
        self.cache = {}
        self.lock = threading.Lock()
    def get(self, key):
        with self.lock:
            return self.cache.get(key)
    def set(self, key, value):
        with self.lock:
            self.cache[key] = value
    def update_if_absent(self, key, compute_func):
        """如果缓存中没有，则计算并写入"""
        with self.lock:
            if key not in self.cache:
                self.cache[key] = compute_func()
            return self.cache[key]
# 使用示例
cache = ThreadSafeCache()
def expensive_operation(key):
    time.sleep(1)  # 模拟耗时操作
    return f"result_{key}"
# 在多个线程中使用
def worker(key):
    result = cache.update_if_absent(key, lambda: expensive_operation(key))
    print(f"Got result: {result}")

使用 multiprocessing.Lock（进程安全）

import multiprocessing
import time
class ProcessSafeCache:
    def __init__(self):
        self.manager = multiprocessing.Manager()
        self.cache = self.manager.dict()
        self.lock = multiprocessing.Lock()
    def get(self, key):
        with self.lock:
            return self.cache.get(key)
    def set(self, key, value):
        with self.lock:
            self.cache[key] = value
def worker(cache, key):
    # 双重检查锁（Double-checked locking）
    result = cache.get(key)
    if result is None:
        with cache.lock:
            # 再次检查，防止重复计算
            result = cache.get(key)
            if result is None:
                time.sleep(1)  # 模拟耗时操作
                result = f"result_{key}"
                cache.set(key, result)
    print(f"Process got: {result}")
if __name__ == '__main__':
    cache = ProcessSafeCache()
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(cache, "key1"))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()

使用 asyncio.Lock（异步安全）

import asyncio
from functools import lru_cache
class AsyncCache:
    def __init__(self):
        self.cache = {}
        self.lock = asyncio.Lock()
    async def get(self, key):
        async with self.lock:
            return self.cache.get(key)
    async def set(self, key, value):
        async with self.lock:
            self.cache[key] = value
    async def get_or_compute(self, key, compute_func):
        async with self.lock:
            if key not in self.cache:
                self.cache[key] = await compute_func()
            return self.cache[key]
async def expensive_async_operation(key):
    await asyncio.sleep(1)  # 模拟异步耗时操作
    return f"async_result_{key}"
async def main():
    cache = AsyncCache()
    # 并发执行多个协程
    tasks = [
        cache.get_or_compute("key1", lambda: expensive_async_operation("key1"))
        for _ in range(3)
    ]
    results = await asyncio.gather(*tasks)
    print(f"Results: {results}")
asyncio.run(main())

使用 Redis 实现分布式锁（跨进程/跨机器）

import redis
import time
import uuid
class RedisDistributedCache:
    def __init__(self, redis_client, lock_timeout=10):
        self.redis = redis_client
        self.lock_timeout = lock_timeout
    def acquire_lock(self, lock_name, acquire_timeout=5):
        """获取分布式锁"""
        identifier = str(uuid.uuid4())
        end = time.time() + acquire_timeout
        while time.time() < end:
            if self.redis.setnx(f"lock:{lock_name}", identifier):
                self.redis.expire(f"lock:{lock_name}", self.lock_timeout)
                return identifier
            time.sleep(0.001)
        return None
    def release_lock(self, lock_name, identifier):
        """释放分布式锁"""
        # 使用 Lua 脚本确保原子操作
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(lua_script, 1, f"lock:{lock_name}", identifier)
    def get_or_compute(self, key, compute_func, ttl=3600):
        """获取缓存或计算"""
        # 先尝试读取缓存
        cached = self.redis.get(key)
        if cached:
            return cached.decode()
        # 获取锁
        lock_id = self.acquire_lock(key)
        if lock_id:
            try:
                # 双重检查
                cached = self.redis.get(key)
                if cached:
                    return cached.decode()
                # 计算并缓存
                result = compute_func()
                self.redis.setex(key, ttl, result)
                return result
            finally:
                self.release_lock(key, lock_id)
        else:
            # 无法获取锁，等待或重试
            time.sleep(0.1)
            return self.get_or_compute(key, compute_func, ttl)
# 使用示例
redis_client = redis.Redis(host='localhost', port=6379, db=0)
distributed_cache = RedisDistributedCache(redis_client)
def expensive_computation(x):
    time.sleep(2)
    return x * x
# 模拟并发调用
result = distributed_cache.get_or_compute("calc_5", lambda: expensive_computation(5))
print(f"Result: {result}")

使用 cachetools 库（推荐）

from cachetools import cached, TTLCache
import threading
import time
# 创建线程安全的 TTL 缓存
cache = TTLCache(maxsize=100, ttl=300)  # 300秒过期
@cached(cache, lock=threading.Lock())
def expensive_function(key):
    """自动处理并发的缓存读写"""
    time.sleep(1)  # 模拟耗时操作
    return f"result_{key}"
# 多线程并发调用
def worker(key):
    result = expensive_function(key)
    print(f"Thread got: {result}")
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=("test_key",))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

最佳实践建议

选择合适的锁类型：
- 单进程多线程：threading.Lock
- 多进程：multiprocessing.Lock 或 Redis
- 异步IO：asyncio.Lock
- 分布式系统：Redis 或 Zookeeper
使用双重检查锁：减少锁竞争，提高性能
设置合理的超时时间：防止死锁
考虑缓存雪崩/穿透：
- 使用布隆过滤器
- 设置随机过期时间
- 限流降级
监控和告警：添加缓存命中率、锁等待时间等监控指标

选择哪种方案取决于你的具体场景：并发级别、数据一致性要求、系统架构等因素。