1、redis-py不需要显式使用连接池。
在幕后，redispy使用一个连接池来管理与Redis服务器的连接。默认情况下，每个Redis实例将依次创建自己的连接池。您可以通过将已创建的连接池实例传递给Redis类的connection_pool参数，来重写此行为并使用现有的连接池。您可以选择这样做，以便实现客户端分片或对连接的管理方式有更好的粒度控制。
2、一般写 r = Redis('127.0.0.1','6379'),别的模块导入r属性，一直保持使用这一个连接对象就可以了，不需要使用连接池，直接单例就可以。如果想无限制实例化使用同一个连接池对象，使用连接池，用法如下。
pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
>>> r1 = redis.Redis(connection_pool=pool)
>>> r2 = redis.Redis(connection_pool=pool)
类似于这样。 控制一下，每个redis服务器只有一个连接池。
# coding=utf8
import redis
import unittest class RedisManager(object):
_pool_dict = {} def __init__(self, host='127.0.0.1', port=6379, db=0, password=''):
if (host, port, db, password) not in self.__class__._pool_dict:
print '创建一个连接池'
self.__class__._pool_dict[(host, port, db, password)] = redis.ConnectionPool(host=host, port=port, db=db, password=password)
self._r = redis.Redis(connection_pool=self._pool_dict[(host, port, db, password)])
self._ping() def get_redis(self):
"""
:rtype :redis.Redis
"""
return self._r def _ping(self):
try:
self._r.ping()
except Exception as e:
raise e class _Test(unittest.TestCase):
def test_error_conn(self):
"""测试错误的连接"""
self.assertRaises(Exception, RedisManager, password='') def test_get_redis(self):
"""正常的连接,测试连接池创建次数"""
RedisManager().get_redis()
RedisManager().get_redis()
RedisManager().get_redis()
RedisManager().get_redis()
r = RedisManager().get_redis()
self.assertIsInstance(r, redis.Redis) def test_set_key_value(self):
"""设置值"""
r = RedisManager().get_redis()
r.set('a', 'c')
self.assertEqual(r.get('a'), 'c', msg='a的值不是c') if __name__ == '__main__':
unittest.main()