異步下 redis的配置
from functools import wraps
from typing import Optional
from redis import asyncio as aioredis
from django.conf import settings
class AsyncRedisClient:
_instance: Optional["AsyncRedisClient"] = None
_redis: Optional[aioredis.Redis] = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if self._redis is None:
self._redis = aioredis.Redis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DB,
password=settings.REDIS_PASSWORD,
username=settings.REDIS_USER_NAME,
decode_responses=True,
# 配置連接池
max_connections=10,
socket_timeout=5,
socket_connect_timeout=5,
retry_on_timeout=True,
health_check_interval=30,
)
async def __aenter__(self):
return self._redis
async def __aexit__(self, exc_type, exc_val, exc_tb):
if exc_type:
# 發生異常時記錄日志
print(f"Redis operation failed: {exc_val}")
return False
@property
def client(self) -> aioredis.Redis:
return self._redis
def get_redis_client() -> aioredis.Redis:
return AsyncRedisClient().client
def with_redis_client(func):
@wraps(func)
async def wrapper(*args, **kwargs):
async with AsyncRedisClient() as redis:
return await func(redis, *args, **kwargs)
return wrapper
# 導出單例實例供直接使用
aio_cache = get_redis_client()
浙公網安備 33010602011771號