直接上代码:
import redis, traceback, sys
from loguru import logger
class RedisClient:
__instance = None
def __new__(cls, *args, **kwargs):
if not cls.__instance:
return object.__new__(cls)
return cls.__instance
def __init__(self, host, port, password=None):
try:
# 拿到一个Redis实例的连接池,避免每次建立、释放连接的开销,节省了每次连接用的时间
self.POLL = redis.ConnectionPool(host=host,
port=port,
decode_responses=True,
db=0,
password=password,
max_connections=100)
logger.info(f'获取Redis连接池, Host={host}, Port={port}')
except Exception as e:
logger.error(f'获取Redis连接池异常, 程序退出:{str(e)},traceback={traceback.format_exc()}')
sys.exit(0)
def get_redis_client(self):
try:
# 从连接池中获取一个连接实例
redis_conn = redis.StrictRedis(connection_pool=self.POLL)
if redis_conn.ping():
logger.info(f'获取Redis连接实例成功')
return redis_conn
except Exception as e:
logger.error(f'Redis连接异常:{str(e)},traceback={traceback.format_exc()}')