python连接redis哨兵_Python redis.sentinel方法代码示例
本文整理汇总了Python中redis.sentinel方法的典型用法代码示例。如果您正苦于以下问题:Python redis.sentinel方法的具体用法?Python redis.sentinel怎么用?Python redis.sentinel使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块redis的用法示例。
在下文中一共展示了redis.sentinel方法的14个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _get_service_names
点赞 6
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def _get_service_names(self): # type: () -> List[six.text_type]
"""
Get a list of service names from Sentinel. Tries Sentinel hosts until one succeeds; if none succeed,
raises a ConnectionError.
:return: the list of service names from Sentinel.
"""
master_info = None
connection_errors = []
for sentinel in self._sentinel.sentinels:
# Unfortunately, redis.sentinel.Sentinel does not support sentinel_masters, so we have to step
# through all of its connections manually
try:
master_info = sentinel.sentinel_masters()
break
except (redis.ConnectionError, redis.TimeoutError) as e:
connection_errors.append('Failed to connect to {} due to error: "{}".'.format(sentinel, e))
continue
if master_info is None:
raise redis.ConnectionError(
'Could not get master info from Sentinel\n{}:'.format('\n'.join(connection_errors))
)
return list(master_info.keys())
开发者ID:eventbrite,项目名称:pysoa,代码行数:25,
示例2: _get_connection
点赞 6
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def _get_connection(self, index): # type: (int) -> redis.StrictRedis
if not 0 <= index < self._ring_size:
raise ValueError(
'There are only {count} hosts, but you asked for connection {index}.'.format(
count=self._ring_size,
index=index,
)
)
for i in range(self._sentinel_failover_retries + 1):
try:
return self._get_master_client_for(self._services[index])
except redis.sentinel.MasterNotFoundError:
self.reset_clients() # make sure we reach out to get master info again on next call
_logger.warning('Redis master not found, so resetting clients (failover?)')
if i != self._sentinel_failover_retries:
self._get_counter('backend.sentinel.master_not_found_retry').increment()
time.sleep((2 ** i + random.random()) / 4.0)
raise CannotGetConnectionError('Master not found; gave up reloading master info after failover.')
开发者ID:eventbrite,项目名称:pysoa,代码行数:22,
示例3: setUp
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def setUp(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.recorder.clear_spans()
# self.sentinel = Sentinel([(testenv['redis_host'], 26379)], socket_timeout=0.1)
# self.sentinel_master = self.sentinel.discover_master('mymaster')
# self.client = redis.Redis(host=self.sentinel_master[0])
self.client = redis.Redis(host=testenv['redis_host'])
开发者ID:instana,项目名称:python-sensor,代码行数:12,
示例4: setUp
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def setUp(self):
pymemcache.client.Client(('localhost', 22122)).flush_all()
redis.from_url('unix:///tmp/limits.redis.sock').flushall()
redis.from_url("redis://localhost:7379").flushall()
redis.from_url("redis://:sekret@localhost:7389").flushall()
redis.sentinel.Sentinel([
("localhost", 26379)
]).master_for("localhost-redis-sentinel").flushall()
rediscluster.RedisCluster("localhost", 7000).flushall()
if RUN_GAE:
from google.appengine.ext import testbed
tb = testbed.Testbed()
tb.activate()
tb.init_memcache_stub()
开发者ID:alisaifee,项目名称:limits,代码行数:16,
示例5: test_storage_check
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def test_storage_check(self):
self.assertTrue(
storage_from_string("memory://").check()
)
self.assertTrue(
storage_from_string("redis://localhost:7379").check()
)
self.assertTrue(
storage_from_string("redis://:sekret@localhost:7389").check()
)
self.assertTrue(
storage_from_string(
"redis+unix:///tmp/limits.redis.sock"
).check()
)
self.assertTrue(
storage_from_string("memcached://localhost:22122").check()
)
self.assertTrue(
storage_from_string(
"memcached://localhost:22122,localhost:22123"
).check()
)
self.assertTrue(
storage_from_string(
"memcached:///tmp/limits.memcached.sock"
).check()
)
self.assertTrue(
storage_from_string(
"redis+sentinel://localhost:26379",
service_name="localhost-redis-sentinel"
).check()
)
self.assertTrue(
storage_from_string("redis+cluster://localhost:7000").check()
)
if RUN_GAE:
self.assertTrue(storage_from_string("gaememcached://").check())
开发者ID:alisaifee,项目名称:limits,代码行数:41,
示例6: setUp
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def setUp(self):
self.storage_url = 'redis+sentinel://localhost:26379'
self.service_name = 'localhost-redis-sentinel'
self.storage = RedisSentinelStorage(
self.storage_url,
service_name=self.service_name
)
redis.sentinel.Sentinel([
("localhost", 26379)
]).master_for(self.service_name).flushall()
开发者ID:alisaifee,项目名称:limits,代码行数:12,
示例7: setUp
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def setUp(self):
pymemcache.client.Client(('localhost', 22122)).flush_all()
redis.from_url("redis://localhost:7379").flushall()
redis.from_url("redis://:sekret@localhost:7389").flushall()
redis.sentinel.Sentinel([
("localhost", 26379)
]).master_for("localhost-redis-sentinel").flushall()
rediscluster.RedisCluster("localhost", 7000).flushall()
开发者ID:alisaifee,项目名称:limits,代码行数:10,
示例8: test_fixed_window_with_elastic_expiry_redis_sentinel
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def test_fixed_window_with_elastic_expiry_redis_sentinel(self):
storage = RedisSentinelStorage(
"redis+sentinel://localhost:26379",
service_name="localhost-redis-sentinel"
)
limiter = FixedWindowElasticExpiryRateLimiter(storage)
limit = RateLimitItemPerSecond(10, 2)
self.assertTrue(all([limiter.hit(limit) for _ in range(0, 10)]))
time.sleep(1)
self.assertFalse(limiter.hit(limit))
time.sleep(1)
self.assertFalse(limiter.hit(limit))
self.assertEqual(limiter.get_window_stats(limit)[1], 0)
开发者ID:alisaifee,项目名称:limits,代码行数:15,
示例9: get_connection
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def get_connection(self, is_read_only=False) -> redis.StrictRedis:
"""
Gets a StrictRedis connection for normal redis or for redis sentinel
based upon redis mode in configuration.
:type is_read_only: bool
:param is_read_only: In case of redis sentinel, it returns connection
to slave
:return: Returns a StrictRedis connection
"""
if self.connection is not None:
return self.connection
if self.is_sentinel:
kwargs = dict()
if self.password:
kwargs["password"] = self.password
sentinel = Sentinel([(self.host, self.port)], **kwargs)
if is_read_only:
connection = sentinel.slave_for(self.sentinel_service,
decode_responses=True)
else:
connection = sentinel.master_for(self.sentinel_service,
decode_responses=True)
else:
connection = redis.StrictRedis(host=self.host, port=self.port,
decode_responses=True,
password=self.password)
self.connection = connection
return connection
开发者ID:biplap-sarkar,项目名称:pylimit,代码行数:33,
示例10: get_atomic_connection
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def get_atomic_connection(self) -> Pipeline:
"""
Gets a Pipeline for normal redis or for redis sentinel based upon
redis mode in configuration
:return: Returns a Pipeline object
"""
return self.get_connection().pipeline(True)
开发者ID:biplap-sarkar,项目名称:pylimit,代码行数:10,
示例11: __init__
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def __init__(self):
if self.REDIS_SENTINEL:
sentinels = [tuple(s.split(':')) for s in self.REDIS_SENTINEL.split(';')]
self._sentinel = redis.sentinel.Sentinel(sentinels,
db=self.REDIS_SENTINEL_DB,
socket_timeout=0.1)
else:
self._redis = redis.Redis.from_url(self.rewrite_redis_url())
开发者ID:ybrs,项目名称:single-beat,代码行数:10,
示例12: _get_master_client_for
点赞 5
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def _get_master_client_for(self, service_name): # type: (six.text_type) -> redis.StrictRedis
if service_name not in self._master_clients:
self._get_counter('backend.sentinel.populate_master_client').increment()
self._master_clients[service_name] = self._sentinel.master_for(service_name)
master_address = self._master_clients[service_name].connection_pool.get_master_address()
_logger.info('Sentinel master address: {}'.format(master_address))
return self._master_clients[service_name]
开发者ID:eventbrite,项目名称:pysoa,代码行数:10,
示例13: _get_redis_connection
点赞 4
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def _get_redis_connection(self, group, shard):
"""
Create and return a Redis Connection for the given group
Returns:
redis.StrictRedis: The Redis Connection
Raises:
Exception: Passes through any exceptions that happen in trying to get the connection pool
"""
redis_group = self.__config.redis_urls_by_group[group][shard]
self.__logger.info(u'Attempting to connect to Redis for group "{}", shard "{}", url "{}"'.format(group, shard,
redis_group))
if isinstance(redis_group, PanoptesRedisConnectionConfiguration):
redis_pool = redis.BlockingConnectionPool(host=redis_group.host,
port=redis_group.port,
db=redis_group.db,
password=redis_group.password)
redis_connection = redis.StrictRedis(connection_pool=redis_pool)
elif isinstance(redis_group, PanoptesRedisSentinelConnectionConfiguration):
sentinels = [(sentinel.host, sentinel.port) for sentinel in redis_group.sentinels]
self.__logger.info(u'Querying Redis Sentinels "{}" for group "{}", shard "{}"'.format(repr(redis_group),
group, shard))
sentinel = redis.sentinel.Sentinel(sentinels)
master = sentinel.discover_master(redis_group.master_name)
password_present = u'yes' if redis_group.master_password else u'no'
self.__logger.info(u'Going to connect to master "{}" ({}:{}, password: {}) for group "{}", shard "{}""'
.format(redis_group.master_name, master[0], master[1], password_present, group, shard))
redis_connection = sentinel.master_for(redis_group.master_name, password=redis_group.master_password)
else:
self.__logger.info(u'Unknown Redis configuration object type: {}'.format(type(redis_group)))
return
self.__logger.info(u'Successfully connected to Redis for group "{}", shard "{}", url "{}"'.format(group,
shard,
redis_group))
return redis_connection
开发者ID:yahoo,项目名称:panoptes,代码行数:46,
示例14: __init__
点赞 4
# 需要导入模块: import redis [as 别名]
# 或者: from redis import sentinel [as 别名]
def __init__(
self,
hosts=None, # type: Optional[Iterable[Union[six.text_type, Tuple[six.text_type, int]]]]
connection_kwargs=None, # type: Dict[six.text_type, Any]
sentinel_services=None, # type: Iterable[six.text_type]
sentinel_failover_retries=0, # type: int
sentinel_kwargs=None, # type: Dict[six.text_type, Any]
):
# type: (...) -> None
# Master client caching
self._master_clients = {} # type: Dict[six.text_type, redis.StrictRedis]
# Master failover behavior
if sentinel_failover_retries < 0:
raise ValueError('sentinel_failover_retries must be >= 0')
self._sentinel_failover_retries = sentinel_failover_retries
connection_kwargs = dict(connection_kwargs) if connection_kwargs else {}
if 'socket_connect_timeout' not in connection_kwargs:
connection_kwargs['socket_connect_timeout'] = 5.0 # so that we don't wait indefinitely during failover
if 'socket_keepalive' not in connection_kwargs:
connection_kwargs['socket_keepalive'] = True
if (
connection_kwargs.pop('ssl', False) or 'ssl_certfile' in connection_kwargs
) and 'connection_class' not in connection_kwargs:
# TODO: Remove this when https://github.com/andymccurdy/redis-py/issues/1306 is released
connection_kwargs['connection_class'] = _SSLSentinelManagedConnection
sentinel_kwargs = dict(sentinel_kwargs) if sentinel_kwargs else {}
if 'socket_connect_timeout' not in sentinel_kwargs:
sentinel_kwargs['socket_connect_timeout'] = 5.0 # so that we don't wait indefinitely connecting to Sentinel
if 'socket_timeout' not in sentinel_kwargs:
sentinel_kwargs['socket_timeout'] = 5.0 # so that we don't wait indefinitely if a Sentinel goes down
if 'socket_keepalive' not in sentinel_kwargs:
sentinel_kwargs['socket_keepalive'] = True
self._sentinel = redis.sentinel.Sentinel(
self._convert_hosts(hosts),
sentinel_kwargs=sentinel_kwargs,
**connection_kwargs
)
if sentinel_services:
self._validate_service_names(sentinel_services)
self._services = list(sentinel_services) # type: List[six.text_type]
else:
self._services = self._get_service_names()
super(SentinelRedisClient, self).__init__(ring_size=len(self._services))
开发者ID:eventbrite,项目名称:pysoa,代码行数:50,
注:本文中的redis.sentinel方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。
python连接redis哨兵_Python redis.sentinel方法代码示例相关推荐
- python程序异常实例_Python werkzeug.exceptions方法代码示例
本文整理汇总了Python中werkzeug.exceptions方法的典型用法代码示例.如果您正苦于以下问题:Python werkzeug.exceptions方法的具体用法?Python wer ...
- python中geometry用法_Python geometry.Point方法代码示例
本文整理汇总了Python中shapely.geometry.Point方法的典型用法代码示例.如果您正苦于以下问题:Python geometry.Point方法的具体用法?Python geome ...
- python re 简单实例_Python re.search方法代码示例
本文整理汇总了Python中re.search方法的典型用法代码示例.如果您正苦于以下问题:Python re.search方法的具体用法?Python re.search怎么用?Python re. ...
- python中config命令_Python config.config方法代码示例
本文整理汇总了Python中config.config方法的典型用法代码示例.如果您正苦于以下问题:Python config.config方法的具体用法?Python config.config怎么 ...
- python中fact用法_Python covariance.EllipticEnvelope方法代码示例
本文整理汇总了Python中sklearn.covariance.EllipticEnvelope方法的典型用法代码示例.如果您正苦于以下问题:Python covariance.EllipticEn ...
- python 求 gamma 分布_Python stats.gamma方法代码示例
本文整理汇总了Python中scipy.stats.gamma方法的典型用法代码示例.如果您正苦于以下问题:Python stats.gamma方法的具体用法?Python stats.gamma怎么 ...
- python messagebox弹窗退出_Python messagebox.showinfo方法代码示例
本文整理汇总了Python中tkinter.messagebox.showinfo方法的典型用法代码示例.如果您正苦于以下问题:Python messagebox.showinfo方法的具体用法?Py ...
- python关于messagebox题目_Python messagebox.askokcancel方法代码示例
本文整理汇总了Python中tkinter.messagebox.askokcancel方法的典型用法代码示例.如果您正苦于以下问题:Python messagebox.askokcancel方法的具 ...
- python安装mlab库_Python mlab.normpdf方法代码示例
本文整理汇总了Python中matplotlib.mlab.normpdf方法的典型用法代码示例.如果您正苦于以下问题:Python mlab.normpdf方法的具体用法?Python mlab.n ...
- python的mag模块_Python mlab.specgram方法代码示例
本文整理汇总了Python中matplotlib.mlab.specgram方法的典型用法代码示例.如果您正苦于以下问题:Python mlab.specgram方法的具体用法?Python mlab ...
最新文章
- redis mysql查询数据类型_linux 常见的标识与Redis数据库详解
- 那天,我被拉入一个Redis群聊···
- 炉石传说服务器维护有补偿吗,炉石传说官网维护补偿什么时候到 未到原因说明...
- 首届世界CSS设计大赛结果揭晓
- Android:图片加载库Glide VS Picasso
- (Oracle学习笔记) Oracle体系结构
- VMware安装系统时没有弹出分区设置
- 一文读懂架构整洁之道(附知识脉络图)
- 北师大计算机学院调剂,北师大数学科学学院2020年硕士研究生调剂方案
- oracle通过dblink连接mysql配置详解(全Windows下)
- 多线程—— Queue(储存进程结果)
- C++queue队列与stack栈
- scrollLeft/scrollTop,offsetLeft/offsetTop,clientLeft/clientTop
- java设置窗口图标
- FLV格式的视频歌曲地址600首,复制地址可插入外链播放器专用
- 用iPad开发iPhone App,苹果发布Swift Playgrounds 4
- 《Photoshop修饰与合成专业技法》目录—导读
- python利用WMI等监控获取windows状态如CPU、内存、硬盘等信息
- 问题 B: Little Sub and Triples
- 利用阿里云ECS制作个人简历网站
热门文章
- ajax清请求过程,JS深入基础之Ajax的请求过程
- 商业方向的大数据专业_结合当前的人才需求趋势,大数据专业考研时可以选择哪些主攻方向...
- 述职答辩提问环节一般可以问些什么_论文答辩一般会问什么问题?需要注意什么事项?...
- SpringBoot聚合项目总结
- vue 单文件组件中,输入template 按 tab 键不能自动补全标签的解决办法
- JavaScript从入门到放弃 -(六)正则表达式
- python numpy sum函数,numpy.sum()的使用详解
- matlab 求n 的和,MATLAB求1的阶乘加到n的阶乘和 不要现有的函数,要自己编写出来的...
- mysql报警代码183_mysql启动报错:/usr/bin/mysqld_safe: line 183: 23716 Killed
- 斯皮尔 皮尔森 肯德尔_一起来学应用统计学(全部)(二)持续更新