Python脚本如何适配高并发缓存读写场景

本文目录导读：

1. 使用高性能连接池
2. 采用异步非阻塞IO
3. 实现缓存穿透/雪崩防护
4. 热点数据保护
5. 使用Pipeline和批量操作
6. 分布式锁实现
7. 合理的序列化策略
8. 监控与调优
9. 完整适配示例

针对Python在高并发缓存读写场景的适配，需要从客户端连接管理、缓存策略、数据结构优化以及异步IO等多个层面进行设计,以下是具体的适配方案。

使用高性能连接池

避免每次请求都创建新连接,使用连接池复用长连接。

import redis
from redis import ConnectionPool
# 创建连接池（全局单例）
pool = ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=50,  # 根据并发量调整
    socket_connect_timeout=2,
    socket_timeout=2,
    retry_on_timeout=True
)
# 从连接池获取连接
r = redis.Redis(connection_pool=pool)

采用异步非阻塞IO

高并发场景下建议使用asyncio + aioredis或redis-py的异步模式。

import asyncio
import aioredis
async def get_cache(key):
    redis = await aioredis.from_url(
        "redis://localhost",
        max_connections=50,
        decode_responses=True
    )
    return await redis.get(key)
async def set_cache(key, value, expire=60):
    redis = await aioredis.from_url(
        "redis://localhost",
        max_connections=50
    )
    await redis.setex(key, expire, value)
# 使用asyncio.gather处理并发
async def handle_requests(keys):
    tasks = [get_cache(key) for key in keys]
    results = await asyncio.gather(*tasks)
    return results

实现缓存穿透/雪崩防护

布隆过滤器（防止缓存穿透）

from pybloom_live import BloomFilter
# 初始化布隆过滤器（全局）
bloom = BloomFilter(capacity=1000000, error_rate=0.001)
# 预热数据
def warmup_bloom():
    for item in all_valid_keys:
        bloom.add(item)
# 查询时先检查
def get_with_bloom(key):
    if key not in bloom:
        return None  # 直接返回，不查询Redis
    return cache_get(key)

缓存空值（防止穿透）

def get_data(key):
    # 尝试从缓存获取
    data = r.get(key)
    if data is not None:
        return data
    # 缓存未命中，从数据库加载
    data = db.query(key)
    if data is None:
        # 缓存空值，过期时间短（如30秒）
        r.setex(key, 30, "NULL")
        return None
    # 缓存实际数据
    r.setex(key, 3600, data)
    return data

热点数据保护

使用互斥锁（防止缓存击穿）

import threading
class HotKeyMutex:
    def __init__(self):
        self._locks = {}
        self._lock = threading.Lock()
    def acquire(self, key):
        with self._lock:
            if key not in self._locks:
                self._locks[key] = threading.Lock()
            return self._locks[key]
    def release(self, key):
        with self._lock:
            if key in self._locks:
                del self._locks[key]
def get_hot_data(key):
    # 先尝试快速获取
    data = r.get(key)
    if data:
        return data
    # 获取锁，防止多个请求同时回源
    lock = mutex.acquire(key)
    with lock:
        # 双重检查
        data = r.get(key)
        if data:
            return data
        # 回源到数据库
        data = db.query(key)
        r.setex(key, 3600, data)
        return data
    finally:
        mutex.release(key)

使用Pipeline和批量操作

减少网络往返次数：

def batch_set(items):
    """items: [(key, value), ...]"""
    pipe = r.pipeline()
    for key, value in items:
        pipe.setex(key, 3600, value)
    pipe.execute()
def batch_get(keys):
    pipe = r.pipeline()
    for key in keys:
        pipe.get(key)
    return pipe.execute()

分布式锁实现

import redis
from redis.lock import Lock
class DistributedLock:
    def __init__(self, redis_client):
        self.redis = redis_client
    def acquire(self, key, timeout=5):
        lock = self.redis.lock(
            f"lock:{key}",
            timeout=timeout,
            blocking_timeout=1  # 等待锁时间
        )
        return lock
# 使用示例
lock = distributed_lock.acquire("hot_data")
try:
    if lock.acquire():
        # 处理业务逻辑
        pass
finally:
    lock.release()

合理的序列化策略

import pickle
import json
import msgpack
# 选择高效的序列化方式
serializer = {
    'json': lambda x: json.dumps(x).encode(),
    'pickle': lambda x: pickle.dumps(x),
    'msgpack': lambda x: msgpack.packb(x)
}
def cache_set(key, value, expire=3600):
    # 推荐msgpack，性能最优
    serialized = serializer['msgpack'](value)
    r.setex(key, expire, serialized)
def cache_get(key):
    data = r.get(key)
    if data:
        return msgpack.unpackb(data)
    return None

监控与调优

import time
from functools import wraps
def monitor_cache(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            # 记录到监控系统
            if elapsed > 0.1:  # 超过100ms报警
                print(f"WARNING: {func.__name__} took {elapsed*1000:.2f}ms")
            return result
        except Exception as e:
            print(f"Cache error: {e}")
            return None
    return wrapper
@monitor_cache
def get_user_info(user_id):
    return r.get(f"user:{user_id}")

完整适配示例

import asyncio
import aioredis
from typing import Optional, Any
class HighConcurrencyCache:
    def __init__(self, redis_url: str, max_connections: int = 50):
        self.redis_url = redis_url
        self.max_connections = max_connections
        self._redis: Optional[aioredis.Redis] = None
        self._bloom = BloomFilter(capacity=1000000, error_rate=0.001)
        self._mutex = HotKeyMutex()
    async def get_redis(self) -> aioredis.Redis:
        if not self._redis:
            self._redis = await aioredis.from_url(
                self.redis_url,
                max_connections=self.max_connections,
                decode_responses=True
            )
        return self._redis
    async def get(self, key: str, expire: int = 3600) -> Any:
        redis = await self.get_redis()
        # 布隆过滤器检查
        if key not in self._bloom:
            return None
        # 快速获取
        data = await redis.get(key)
        if data:
            return data
        # 互斥锁防止击穿
        async with self._mutex.acquire(key):
            # 双重检查
            data = await redis.get(key)
            if data:
                return data
            # 模拟数据库查询
            data = await self._load_from_db(key)
            if data:
                await redis.setex(key, expire, data)
                self._bloom.add(key)
            else:
                # 缓存空值
                await redis.setex(key, 30, "NULL")
            return data
    async def _load_from_db(self, key: str) -> Any:
        # 实际数据库查询逻辑
        await asyncio.sleep(0.01)  # 模拟IO
        return f"data_{key}"
# 使用示例
async def main():
    cache = HighConcurrencyCache("redis://localhost")
    # 并发1000个请求
    tasks = [cache.get(f"key_{i}") for i in range(1000)]
    results = await asyncio.gather(*tasks)
    print(f"Processed {len(results)} requests")
asyncio.run(main())

根据实际业务场景选择合适的组合方案，通常建议异步IO + 连接池 + 布隆过滤器 + 互斥锁作为基础配置。