1、redis-py不需要顯式使用連接池。

在幕后，redispy使用一個連接池來管理與Redis服務器的連接。默認情況下，每個Redis實例將依次創(chuàng)建自己的連接池。您可以通過將已創(chuàng)建的連接池實例傳遞給Redis類的connection_pool參數(shù)，來重寫此行為并使用現(xiàn)有的連接池。您可以選擇這樣做，以便實現(xiàn)客戶端分片或?qū)B接的管理方式有更好的粒度控制。

?

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 unittestclass RedisManager(object):_pool_dict = {}def __init__(self, host='127.0.0.1', port=6379, db=0, password='123456'):if (host, port, db, password) not in self.__class__._pool_dict:print '創(chuàng)建一個連接池'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._rdef _ping(self):try:self._r.ping()except Exception as e:raise eclass _Test(unittest.TestCase):def test_error_conn(self):"""測試錯誤的連接"""self.assertRaises(Exception, RedisManager, password='1234567')def test_get_redis(self):"""正常的連接,測試連接池創(chuàng)建次數(shù)"""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()

?

? ?

轉(zhuǎn)載于:https://www.cnblogs.com/ydf0509/p/9070084.html

總結(jié)

以上是生活随笔為你收集整理的关于redis连接池的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。