关于redis连接池
生活随笔
收集整理的這篇文章主要介紹了
关于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 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 '創建一個連接池'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):"""正常的連接,測試連接池創建次數"""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()
?
? ?轉載于:https://www.cnblogs.com/ydf0509/p/9070084.html
總結
以上是生活随笔為你收集整理的关于redis连接池的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 前端开发之模块化初探
- 下一篇: flume + kafka