Python使用redis pool的一种单例实现方式_Python

Python使用redis pool的一种单例实现方式

2020-08-19 10:21mo_guang Python

这篇文章主要介绍了Python使用redis pool的一种单例实现方式,结合实例形式分析了Python操作redis模块实现共享同一个连接池的相关技巧,需要的朋友可以参考下

本文实例讲述了Python使用redis pool的一种单例实现方式。分享给大家供大家参考，具体如下：

为适应多个redis实例共享同一个连接池的场景，可以类似于以下单例方式实现:

									import redis

									class RedisDBConfig:

									  HOST = '127.0.0.1'

									  PORT = 6379

									  DBID = 0

									def operator_status(func):

									  '''''get operatoration status

									  '''

									  def gen_status(*args, **kwargs):

									    error, result = None, None

									    try:

									      result = func(*args, **kwargs)

									    except Exception as e:

									      error = str(e)

									    return {'result': result, 'error': error}

									  return gen_status

									class RedisCache(object):

									  def __init__(self):

									    if not hasattr(RedisCache, 'pool'):

									      RedisCache.create_pool()

									    self._connection = redis.Redis(connection_pool = RedisCache.pool)

									  @staticmethod

									  def create_pool():

									    RedisCache.pool = redis.ConnectionPool(

									        host = RedisDBConfig.HOST,

									        port = RedisDBConfig.PORT,

									        db  = RedisDBConfig.DBID)

									  @operator_status

									  def set_data(self, key, value):

									    '''''set data with (key, value)

									    '''

									    return self._connection.set(key, value)

									  @operator_status

									  def get_data(self, key):

									    '''''get data by key

									    '''

									    return self._connection.get(key)

									  @operator_status

									  def del_data(self, key):

									    '''''delete cache by key

									    '''

									    return self._connection.delete(key)

									if __name__ == '__main__':

									  print RedisCache().set_data('Testkey', "Simple Test")

									  print RedisCache().get_data('Testkey')

									  print RedisCache().del_data('Testkey')

									  print RedisCache().get_data('Testkey')