REDIS 如何利用python 操作redis 集群 (投稿文章)
注明: 此篇文章为投稿文字, 投稿人 闫树爽, (程序员, 目前从事REDIS ,MONGODB ,以及数据库运维自动化代码工作)
在NOSQL 数据库中的操作,与关系型数据库不同的是,会一门程序对于NOSQL的数据库操控是十分有利的,下面的内容是关于如何利用python程序来操作redis 集群的说明.
## 利用python操作redis集群
redis的cluster模式为大型应用中常用的方式，今天学习如何使用redis-py-cluster来操作redis集群
首先安装redis-py-cluster
```
pip install redis-py-cluster
```
基本用法
```python
from rediscluster import RedisCluster
from string import ascii_letters
import random
conn = RedisCluster(host="127.0.0.1", port=6379, password='password')#创建连接
for i in range(10000):
key = ''.join(random.sample(ascii_letters, k=7))#创建一个随机字符串作为key
# ex = random.randint(1, 60 * 60 * 24)
conn.set(key, i, ex=500)
print(i)
```
此时可以发现reidis中已经有了我们的数据
然而在现实使用中，redis的使用往往涉及高并发，每次都重新创建连接不是我们所建议的方式，我们可以使用连接池来创建连接，并通过多线程来进行访问。
redis-py-cluster的官方文档写的比较简单并没有给出详细的连接池使用方式，但是好在python能够查看源码，我们可以看到其中有一个ClusterConnectionPool类，这个从命名来看应该是连接池的。进源码去看这个类是继承自redis库的ConnectionPool。
直接连接尝试
```python
from rediscluster import ClusterBlockingConnectionPool,RedisCluster,ClusterConnectionPool
startup_nodes = [
{'host': '10.50.132.92', 'port': 6379},
{'host': '10.50.132.93', 'port': 6379},
{'host': '10.50.132.94', 'port': 6379},
]
pool = ClusterConnectionPool(startup_nodes=startup_nodes,password="xxxx")
```
## 处理报错
会发现报错（个别现象，原因下面说），如果未遇到错误可以跳过这小节
```
redis.exceptions.ResponseError: unknown command `CONFIG`, with args beginning with: `GET`, `cluster-require-full-coverage`,
```
原因是因为我将redis中的config命令rename掉了，但是在默认情况下，会检查槽的完整性，所以会使用到config命令，但是config命令已经被我rename掉了啊，所以会报这个错误。解决方法有两种：
1、修改自己的代码，这个参数就是字面意思，跳过完整性检查，这样也就不会调用config命令了。
```python
pool = ClusterBlockingConnectionPool(startup_nodes=startup_nodes,password="xxxx"，skip_full_coverage_check=True)
```
2、修改库的代码，这样的好处是还是能够检查完整行，但是缺点是在进行项目迁移的时候还要再改一遍源码，修改方法如下：
首先找到redis包中的client.py，一般在site-packages/redis/client.py
如果找不到可以从报错的地方快速找到redis包的安装路径
找到这个1239行
```python
def config_get(self, pattern="*"):
"Return a dictionary of configuration based on the ``pattern``"
return self.execute_command('CONFIG GET', pattern)
```
修改这个CONFIG GET为rename后的命令就可以了
通过以上两种方法，可以正常使用连接池了
## 连接池的使用
这个连接池的使用方法如下
```python
from rediscluster import ClusterBlockingConnectionPool,RedisCluster,ClusterConnectionPool
startup_nodes = [
{'host': '10.50.132.92', 'port': 6379},
{'host': '10.50.132.93', 'port': 6379},
{'host': '10.50.132.94', 'port': 6379},
]
pool = ClusterConnectionPool(startup_nodes=startup_nodes,password="xxxx")
client = RedisCluster(connection_pool=pool)
```
这里的pool使用上面的方式就可以得到了一个client
## 实际使用
我们可以用多线程为例子，使用连接池来操作redis
```python
from rediscluster import ClusterBlockingConnectionPool,RedisCluster,ClusterConnectionPool
import threading
from string import ascii_letters
import random
startup_nodes = [
{'host': '10.50.132.92', 'port': 6379},
{'host': '10.50.132.93', 'port': 6379},
{'host': '10.50.132.94', 'port': 6379},
]
pool = ClusterConnectionPool(startup_nodes=startup_nodes,password="xxxx",max_connections=10)
threads=[]
lock = threading.Lock()
def test_thread(thread_id):
client = RedisCluster(connection_pool=pool)
for i in range(100):
key = ''.join(random.sample(ascii_letters, k=7))
client.set(key,random.randint(1,100),ex=100)
lock.acquire()
print(f"Thread-{thread_id}:processed {i+1} times")
lock.release()
#创建线程
for i in range(10):
threads.append(threading.Thread(target=test_thread,args=(i,)))
#开启线程
for th in threads:
th.start()
```
此处我们的线程池里面放10个连接，然后开启10个线程，为了显示的不太难看，我们在print的地方加了个锁，我们期待这个样能够正常的10个线程同时向redis里面写数据，但是实际却发现
会显示有太多的连接，猜测应该是连接池中的连接不够，所以我们调大max_connections参数为50，发现可以正常使用了，但是为什么呢？
熟悉redis的同学可能一下就猜测到了原因。因为我们的集群模式，key键是要根据hash值来分配的，具体连接到那个我们插入之间是不知道的，所以在连接创建之前，客户端也是不知道的，所以显示出这个。在初始化函数中果然找到了相应的参数max_connections_per_node。我们将它设置为True。
这样测试，我们将max_connections设置为10，同时将max_connections_per_node设置为True
这样一来每个节点都有10个连接，我们可以开启30个线程，如下
```python
from rediscluster import ClusterBlockingConnectionPool,RedisCluster,ClusterConnectionPool
import threading
from string import ascii_letters
import random
startup_nodes = [
{'host': '10.50.132.92', 'port': 6379},
{'host': '10.50.132.93', 'port': 6379},
{'host': '10.50.132.94', 'port': 6379},
]
pool = ClusterConnectionPool(startup_nodes=startup_nodes,password="xxx",max_connections=10,skip_full_coverage_check=True,max_connections_per_node=True)
threads=[]
lock = threading.Lock()
def test_thread(thread_id):
client = RedisCluster(connection_pool=pool)
for i in range(100):
key = ''.join(random.sample(ascii_letters, k=7))
client.set(key,random.randint(1,100),ex=100)
lock.acquire()
print(f"Thread-{thread_id}:processed {i+1} times")
lock.release()
#创建线程
for i in range(30):
threads.append(threading.Thread(target=test_thread,args=(i,)))
#开启线程
for th in threads:
th.start()
```
但是~~~，还是不行，跑了一段时间以后有的线程会报错
简单思考下，因为这样虽然开启了这么多的pool，但是对于30个线程来说，仍然有可能引发冲突，因为我们设置的是每个节点10个连接
但是30个线程在同一时间可能有11个访问到了同一个节点，所以会产生这个错误。所以这里应该改成max_connections=30
但是如果我们连接池只想创建20个呢？我们需要自己加锁判断吗？这样是否太麻烦了？
of course not~
在上面大家应该注意到，我在引入的时候还另外引入了一个ClusterBlockingConnectionPool，这个注释中写的是
```
Thread-safe blocking connection pool for Redis Cluster::
```
这是一个线程安全的连接池，继承自ClusterConnectionPool，
使用我们仅需要将ClusterConnectionPool替换为ClusterBlockingConnectionPool即可。
至此已经可以正常使用redis连接池。
## 时间对比：
多线程连接池方式30个线程每个线程插入100个共3000次数据耗时5.1秒
单线程访问插入1000个数据用时49秒，如果3000个应该在150秒左右，大概差了不到30倍，符合我们开的30个线程
## 空闲超时
在没有经过测试之前，我有一个担忧就是服务端设置了空闲超时，我们这个python客户端会被服务端断开，那么单机还好，因为每次是单独的创建连接，那么连接池呢？
我们可以在test_thread中加入一个延时，然后过一段时间让连接空闲并超时以后再访问。经过测试，还是能连接的。
相关文章