大家好~我是
米洛！
我正在從0到1打造一個開源的接口測試平台, 也在編寫一套與之對應的完整教程，希望大家多多支持。
歡迎關注我的公眾號米洛的測開日記，獲取最新文章教程!
回顧
上一節我們編寫了Redis的相關配置編輯頁面，博主這里也趁熱打鐵，把前端頁面完善了。（可能會有一點點小問題，但應該主流程都正常）
其實和其他配置管理頁面差不多，前端優化了一下面包屑，頂部的菜單也放回到左側了。看看mac下的效果:
搜索選項改動了一些，所見即所得，如果搜索項發生變化，那么內容也會隨之切換
關於Redis客戶端的選用
其實在這個問題上我是比較糾結的，redis有star很多的py客戶端，也有與之對應的集群版本。但他們並不支持asyncio。
而支持asyncio的aioredis，本身是個很好的選擇，但人家沒有支持redis集群的計划。orz
所以今天想的是要不就用個同步的redis-cluster-py庫算了，不過在我翻了github一段時間，發現了個叫aredis的異步庫。大概瞅了下，他基本上是保持了和redis-cluster-py接近的api，可能也是為了吸引用戶。
所以咱們就先試驗一下，小白鼠嘛，總得有人來做。
安裝aredis
看官網是要安裝aredis[hiredis]，但我好像不適合這樣方式，於是我分開裝:
pip3 install aredis hiredis
編寫RedisManager
其實這里還是和MySQL比較接近的，也是通過一個字典存放各個redis的連接配置。
不過由於Redis的集群和單實例還有一點區別（好在我們編寫配置的時候就准備好了），所以我們最好是針對單實例和集群分別編寫2個map存放他們的client，當然1個也是ok的。
整體流程: 從字典獲取客戶端，如果沒有則新開一個客戶端，並放入緩存，有則返回。
-
可能存在的問題
代碼不是線程安全的，需要觀察是否需要加鎖
緩存不像LRU會降頻，也不能自動過期
對我來說第一個肯定是個大問題，如果出現了就必須得解決。至於第二個問題，由於redis配置很少變動，而且我們本身是連接池的形式，所以影響不算大。
話不多說，現在我們就來編寫吧:
"""
redis客戶端，基於aredis(支持集群，aioredis不支持集群)
"""
from aredis import StrictRedisCluster, ClusterConnectionPool, ConnectionPool, StrictRedis
from app.excpetions.RedisException import RedisException
class PityRedisManager(object):
"""非線程安全，可能存在問題
"""
_cluster_pool = dict()
_pool = dict()
@staticmethod
def get_cluster_client(redis_id: int, addr: str):
"""
獲取redis集群客戶端
:param redis_id:
:param addr:
:return:
"""
cluster = PityRedisManager._cluster_pool.get(redis_id)
if cluster is not None:
return cluster
client = PityRedisManager.get_cluster(addr)
PityRedisManager._cluster_pool[redis_id] = client
return client
@staticmethod
def get_single_node_client(redis_id: int, addr: str, password: str, db: str):
"""
獲取redis單實例客戶端
:param redis_id:
:param addr:
:param password:
:param db:
:return:
"""
node = PityRedisManager._cluster_pool.get(redis_id)
if node is not None:
return node
host, port = addr.split(":")
pool = ConnectionPool(host=host, port=port, db=db, max_connections=100, password=password,
decode_responses=True)
client = StrictRedis(connection_pool=pool)
PityRedisManager._pool[redis_id] = PityRedisManager.get_cluster(addr)
return client
@staticmethod
def refresh_redis_client(redis_id: int, addr: str, password: str, db: str):
"""
刷新redis客戶端
:param redis_id:
:param addr:
:param password:
:param db:
:return:
"""
host, port = addr.split(":")
pool = ConnectionPool(host=host, port=port, db=db, max_connections=100, password=password,
decode_responses=True)
client = StrictRedis(connection_pool=pool, decode_responses=True)
PityRedisManager._pool[redis_id] = client
@staticmethod
def refresh_redis_cluster(redis_id: int, addr: str):
PityRedisManager._cluster_pool[redis_id] = PityRedisManager.get_cluster(addr)
@staticmethod
def get_cluster(addr: str):
"""
獲取集群連接池
:param addr:
:return:
"""
try:
nodes = addr.split(',')
startup_nodes = [{"host": n.split(":")[0], "port": n.split(":")[1]} for n in nodes]
pool = ClusterConnectionPool(startup_nodes=startup_nodes, max_connections=100, decode_responses=True)
client = StrictRedisCluster(connection_pool=pool, decode_responses=True)
return client
except Exception as e:
raise RedisException(f"獲取Redis連接失敗, {e}")
我們以數據庫的唯一id為key，緩存redis的連接池。
由於連接池會自動開啟/關閉連接，所以我們不需要手動關閉客戶端，非常方便。
可以明顯看到我們分別用了ClusterConnectionPool和ConnectionPool，分別對應集群和實例。參數基本上算是一致。
至於refresh，是給改動redis以后做的刷新連接的工作。
以上就是RedisManager的內容，到這只是能夠獲取Redis客戶端了。
嘗試一下
有條件的同學可以本次安裝redis:
$ wget https://download.redis.io/releases/redis-6.2.6.tar.gz
$ tar xzf redis-6.2.6.tar.gz
$ cd redis-6.2.6
$ make
make了以后，修改redis-6.2.6目錄下的redis.conf, 接着取消這一行的注釋:
使用密碼模式（redis最好是加密碼，端口號也盡量不要用原生的6379，本寶寶有台機器被人通過redis植入了挖礦程序，苦不堪言）
- 在redis-6.2.6目錄下啟動
src/redis-server redis.conf
這樣本地redis的實例就啟動了~
編寫個在線測試redis的接口
-
先通過id拿到redis的配置信息
-
然后通過manager拿到連接池
-
對redis發動命令
我們在http://localhost:7777/docs打開swagger調試:
-
讀取faker
- 設置faker為s12
- 再次取faker
可以看到redis的相關操作已經是可以用了，那我們今天的內容就到這了，愉快的周末總是辣么短暫。
下一節我們就得編寫在線執行Redis的命令及相關頁面了！