关于redis连接池

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()

巴特西

关于redis连接池

最新文章

热门文章