Redis Cluster 部署与运维：向量搜索场景下的分片、故障转移和容量规划

2026/7/9 16:42:26

Redis Cluster 部署与运维：向量搜索场景下的分片、故障转移和容量规划

一、深度引言与场景痛点

大家好，我是赵咕咕。

你的 RAG 系统接了第一个大客户，文档量从 10 万涨到 500 万。单机 Redis 扛不住了——一个 16GB 内存的实例，向量索引就占了 12GB。你再往里面塞数据，OOM 直接挂掉。

摆在你面前的选择：要么买更大的机器（贵），要么上 Redis Cluster（复杂）。很多人选前者，就因为觉得 Cluster "配置起来太麻烦"。

但实际上，对于向量搜索这种内存密集型场景，单机永远有天花板。Redis Cluster 的分片 + 故障转移能力，正是你需要的扩展方案。这篇文章我们不走理论，直接上实战配置和运维策略。

二、底层机制与原理深度剖析

Redis Cluster 使用哈希槽（Hash Slot）机制进行数据分片。总共 16384 个槽，平均分配给各个节点。写入时根据 key 的 CRC16 哈希值决定落在哪个槽，再路由到对应节点。

对于向量搜索，最关键的考虑是：向量检索不能跨节点。因为 HNSW 图结构是节点本地的，跨节点搜索需要客户端做"分散-聚合"：

flowchart TB A[客户端请求<br/>向量搜索 + 过滤条件] --> B[计算 key 的哈希槽] B --> C{key 在哪个节点?} C -->|单个节点| D1[节点 1<br/>本地 ANN 搜索] C -->|广播查询| D[分散查询<br/>发送到所有节点] D --> E1[节点 1<br/>HNSW 局部搜索] D --> E2[节点 2<br/>HNSW 局部搜索] D --> E3[节点 3<br/>HNSW 局部搜索] D1 --> F[返回结果] E1 --> G[客户端聚合<br/>去重 + 重排序] E2 --> G E3 --> G G --> F H[故障检测<br/>Gossip 协议] -.-> D1 H -.-> E1 H -.-> E2 H -.-> E3 I[槽迁移<br/>rebalance] -.-> C style B fill:#fff3e0 style G fill:#e8f5e9 style H fill:#f3e5f5

Cluster 模式下向量搜索的三个关键挑战：

分散-聚合的延迟：查询需要发给所有节点，总延迟 = 最慢节点的响应时间
槽迁移时的数据一致性：扩容时槽在节点间迁移，正在迁移的槽数据在两端都可能存在
故障转移的窗口期：主节点挂掉到从节点接管，期间请求会失败

三、生产级代码实现

下面是 Redis Cluster 在向量搜索场景的运维工具集：

from __future__ import annotations import asyncio import numpy as np import time import json from dataclasses import dataclass, field from typing import Optional import redis.asyncio as aioredis from redis.asyncio.cluster import RedisCluster @dataclass class ClusterNode: """集群节点信息""" node_id: str host: str port: int role: str # "master" | "slave" slots: list[str] = field(default_factory=list) memory_used: int = 0 memory_max: int = 0 connected_clients: int = 0 is_connected: bool = True class RedisClusterManager: """Redis Cluster 运维管理器""" def __init__( self, startup_nodes: list[dict], password: str = "", max_connections: int = 100, ): """ startup_nodes: 初始节点列表 [{"host": "10.0.1.1", "port": 6379}, ...] """ self._startup_nodes = startup_nodes self._password = password self._max_connections = max_connections self._cluster: Optional[RedisCluster] = None async def connect(self) -> bool: """连接到集群""" try: self._cluster = RedisCluster( host=self._startup_nodes[0]["host"], port=self._startup_nodes[0]["port"], password=self._password or None, max_connections=self._max_connections, # 关键：允许 MOVED/ASK 重定向 max_connections_per_node=50, retry_on_timeout=True, retry_on_error=[ConnectionError, TimeoutError], # 连接池配置 socket_keepalive=True, socket_connect_timeout=5, # Redis Cluster 特定配置 read_from_replicas=False, # 向量搜索不建议读从节点 ) await self._cluster.ping() return True except Exception as e: print(f"集群连接失败: {e}") return False async def cluster_info(self) -> dict: """获取集群信息""" if not self._cluster: return {} info = await self._cluster.cluster_info() # 解析状态 result = {"raw": info} for line in info.split("\n"): if ":" in line: key, val = line.split(":", 1) result[key] = val return result async def get_nodes(self) -> list[ClusterNode]: """获取所有节点信息""" if not self._cluster: return [] slots_info = await self._cluster.cluster_slots() nodes = [] seen = set() for slot_range in slots_info: master = slot_range[2] node_key = f"{master[0]}:{master[1]}" if node_key not in seen: seen.add(node_key) # 获取节点内存信息 memory_info = await self._get_node_memory( master[0], master[1] ) nodes.append(ClusterNode( node_id=slot_range[2][2], host=master[0], port=master[1], role="master", slots=[f"{slot_range[0]}-{slot_range[1]}"], **memory_info, )) # 从节点 for replica in slot_range[3:]: r_key = f"{replica[0]}:{replica[1]}" if r_key not in seen: seen.add(r_key) nodes.append(ClusterNode( node_id=replica[2], host=replica[0], port=replica[1], role="slave", )) return nodes async def _get_node_memory( self, host: str, port: int ) -> dict: """获取单节点内存信息""" try: node = aioredis.Redis( host=host, port=port, password=self._password or None, socket_connect_timeout=3, ) info = await node.info("memory") await node.close() return { "memory_used": info.get("used_memory", 0), "memory_max": info.get("maxmemory", 0) or info.get("total_system_memory", 0), } except Exception: return {"memory_used": 0, "memory_max": 0} async def capacity_planning(self) -> dict: """容量规划分析""" nodes = await self.get_nodes() total_memory = sum(n.memory_max for n in nodes) used_memory = sum(n.memory_used for n in nodes) master_count = sum(1 for n in nodes if n.role == "master") usage_pct = ( (used_memory / total_memory * 100) if total_memory > 0 else 0 ) # 估算可容纳的向量数量 # 假设每个向量 1024 维 float32 = 4096 bytes + metadata 1024 bytes bytes_per_vector = 5120 remaining_bytes = total_memory - used_memory estimated_capacity = max(0, remaining_bytes // bytes_per_vector) return { "total_nodes": len(nodes), "master_nodes": master_count, "total_memory_gb": round(total_memory / (1024**3), 2), "used_memory_gb": round(used_memory / (1024**3), 2), "usage_percent": round(usage_pct, 1), "estimated_remaining_vectors": estimated_capacity, "recommendation": self._capacity_recommendation( usage_pct, master_count ), } def _capacity_recommendation( self, usage_pct: float, master_count: int ) -> str: """根据容量使用率给出建议""" if usage_pct > 85: return ( f"紧急：内存使用率 {usage_pct}%，" f"建议立即添加节点或扩容" ) elif usage_pct > 70: return ( f"警告：内存使用率 {usage_pct}%，" f"建议开始规划扩容" ) elif usage_pct > 50: return f"正常：内存使用率 {usage_pct}%，容量充足" else: return ( f"富余：内存使用率 {usage_pct}%，" f"当前 {master_count} 个主节点可能过多" ) async def search_with_cluster( self, query_vector: list[float], top_k: int = 10, index_name: str = "idx:vectors", timeout_ms: float = 500, ) -> list[dict]: """在 Cluster 模式下执行向量搜索""" if not self._cluster: return [] # Cluster 模式下，需要查询所有节点并聚合 # 实际上需要获取所有 master 节点的连接 nodes = await self.get_nodes() masters = [n for n in nodes if n.role == "master"] # 并发向所有主节点发起搜索 tasks = [] for node in masters: task = self._search_node( node, query_vector, top_k, index_name, timeout_ms ) tasks.append(task) all_results = await asyncio.gather(*tasks, return_exceptions=True) # 聚合结果 merged: dict[str, dict] = {} for result_list in all_results: if isinstance(result_list, Exception): continue for doc in result_list: doc_id = doc.get("id", "") if doc_id not in merged or doc["score"] > merged[doc_id]["score"]: merged[doc_id] = doc # 排序并截断 sorted_results = sorted( merged.values(), key=lambda x: x["score"], reverse=True ) return sorted_results[:top_k] async def _search_node( self, node: ClusterNode, query_vector: list[float], top_k: int, index_name: str, timeout_ms: float, ) -> list[dict]: """在单个节点上执行搜索""" try: node_conn = aioredis.Redis( host=node.host, port=node.port, password=self._password or None, socket_connect_timeout=1, socket_timeout=timeout_ms / 1000, ) # 构建向量查询 query_vec = np.array(query_vector, dtype=np.float32).tobytes() from redis.commands.search.query import Query q = ( Query(f"*=>[KNN {top_k * 2} @embedding $vec AS score]") .sort_by("score") .return_fields("id", "content", "score") .dialect(2) ) results = await node_conn.ft(index_name).search( q, query_params={"vec": query_vec} ) await node_conn.close() return [ { "id": doc.id, "content": getattr(doc, "content", ""), "score": float(getattr(doc, "score", 0)), } for doc in results.docs ] except Exception as e: return [] async def close(self) -> None: """关闭连接""" if self._cluster: await self._cluster.close() # 使用示例 async def main(): manager = RedisClusterManager( startup_nodes=[ {"host": "10.0.1.1", "port": 6379}, {"host": "10.0.1.2", "port": 6379}, {"host": "10.0.1.3", "port": 6379}, ], password="your_password", ) # 连接检查 print("连接 Redis Cluster...") # 在实际环境： # connected = await manager.connect() # if connected: # info = await manager.cluster_info() # print(f"集群状态: {info.get('cluster_state')}") # # plan = await manager.capacity_planning() # print(json.dumps(plan, ensure_ascii=False, indent=2)) print("Redis Cluster Manager 已就绪") print("功能: 集群连接、节点管理、容量规划、向量搜索") asyncio.run(main())