博客 / 詳情

返回

艾體寶乾貨 | Redis Python 開發系列#5 高可用與集羣部署指南

本文詳細講解 Redis 主從複製、哨兵模式和集羣模式的原理與配置，提供完整的 Python 連接代碼示例，確保你的應用在生產環境中實現高可用性和可擴展性。

前言

在掌握了 Redis 的所有核心數據結構和高級特性後，我們面臨最後一個關鍵挑戰：如何讓 Redis 在生產環境中不宕機？單點 Redis 實例一旦故障，將導致整個應用不可用。這就是 Redis 高可用（High Availability） 和 集羣（Cluster） 架構要解決的根本問題。

​本篇讀者收益​：

  • 深入理解 Redis 主從複製、哨兵模式（Sentinel） 和 集羣模式（Cluster） 的架構原理。
  • 掌握使用 redis-py 連接 Redis Sentinel 實現自動故障轉移和讀寫分離。
  • 掌握使用 redis-py 或 redis-cluster 連接 Redis Cluster 實現數據分片和水平擴展。
  • 瞭解雲服務（如 AWS ElastiCache、Azure Cache）的連接要點。

​先修要求​：已掌握 Redis 基礎連接和操作（詳見系列前四篇）。

​關鍵要點​：

  1. ​主從複製​：數據冗餘的基礎，從節點提供讀擴展，但不具備自動故障轉移能力。
  2. ​哨兵模式（Sentinel）​：在複製基礎上增加了**監控、通知和自動故障轉移**，實現真正的高可用。
  3. ​集羣模式（Cluster）​：通過數據分片（sharding）實現**水平擴展**，兼具高可用和可擴展性。
  4. 從單機到哨兵再到集羣，是一個在**複雜度、可用性和擴展性**之間的權衡過程。

背景與原理簡述

隨着業務增長，單機 Redis 會遇到兩個核心瓶頸：

  1. ​可用性瓶頸​：單個節點故障導致服務完全中斷。
  2. ​性能/容量瓶頸​：單機內存、CPU、網絡帶寬有限。

Redis 提供了三種進階部署方案來解決這些問題：

  • ​主從複製（Replication）​：一個主節點（master）負責寫操作，多個從節點（slave）複製主節點數據並提供讀服務。解決了**讀擴展**和**數據備份**，但沒有自動故障轉移。
  • ​哨兵模式（Sentinel）​：在複製基礎上，引入專門的哨兵進程來監控節點健康狀態，並在主節點故障時**自動選舉新的主節點**，實現高可用。
  • ​集羣模式（Cluster）​：將數據自動分片到多個主節點上，每個主節點都有對應的從節點。同時解決了**水平擴展**和**高可用**問題。

環境準備與快速上手

安裝必要的 Python 庫

# 安裝 redis-py（支持 Sentinel 和基礎 Cluster 連接）
pip install "redis[hiredis]"

# 對於更完整的 Cluster 支持，推薦安裝 redis-py-cluster
pip install redis-py-cluster

基礎連接測試

# filename: setup.py
import os
import redis
from redis.sentinel import Sentinel
from redis.cluster import RedisCluster
from redis.exceptions import RedisError, ConnectionError

print("Redis 高可用與集羣連接演示環境就緒")

核心用法與代碼示例

主從複製（Replication）

架構概述

  • 一個主節點（可寫可讀）
  • 一個或多個從節點（只讀，異步複製主節點數據）
  • 客户端手動處理讀寫分離

Python 連接示例

# filename: replication_demo.py
def replication_setup():
    """演示主從複製的基本連接（手動讀寫分離）"""
    
    # 連接主節點（寫操作）
    master = redis.Redis(
        host='redis-master-host',
        port=6379,
        password='your_password',
        decode_responses=True
    )
    
    # 連接從節點（讀操作）
    slave = redis.Redis(
        host='redis-slave-host', 
        port=6379,
        password='your_password',
        decode_responses=True
    )
    
    # 寫入只能在主節點
    master.set('global:counter', 100)
    
    # 讀取可以在從節點（注意複製延遲）
    value = slave.get('global:counter')
    print(f"從節點讀取的值: {value}")
    
    return master, slave

# 注意：生產環境不建議手動管理主從，推薦使用 Sentinel

哨兵模式（Sentinel）

架構概述

  • 多個哨兵進程（Sentinel）組成集羣，監控 Redis 節點
  • 自動故障檢測和主節點選舉
  • 客户端通過哨兵發現當前的主節點

Python 連接 Sentinel

# filename: sentinel_demo.py
def sentinel_connection():
    """連接 Redis Sentinel 集羣"""
    
    # 1. 定義哨兵節點列表
    sentinel_nodes = [
        ('sentinel1.example.com', 26379),
        ('sentinel2.example.com', 26379), 
        ('sentinel3.example.com', 26379),
    ]
    
    # 2. 創建 Sentinel 對象
    sentinel = Sentinel(
        sentinel_nodes,
        socket_timeout=0.1,
        password='your_sentinel_password',  # 如果哨兵有密碼
        decode_responses=True
    )
    
    # 3. 獲取主節點和從節點客户端
    # service_name 是在哨兵配置中定義的集羣名稱
    master = sentinel.master_for(
        'my-redis-cluster',  # service_name
        socket_timeout=0.1,
        password='your_redis_password',
        decode_responses=True
    )
    
    slave = sentinel.slave_for(
        'my-redis-cluster',  # service_name  
        socket_timeout=0.1,
        password='your_redis_password',
        decode_responses=True
    )
    
    return master, slave, sentinel

def sentinel_operations():
    """使用 Sentinel 客户端進行操作"""
    try:
        master, slave, sentinel = sentinel_connection()
        
        # 寫入操作 - 使用主節點
        master.set('sentinel:test', 'hello from master')
        master.incr('sentinel:counter')
        
        # 讀取操作 - 使用從節點
        value = slave.get('sentinel:test')
        counter = slave.get('sentinel:counter')
        print(f"從節點讀取: {value}, 計數器: {counter}")
        
        # 發現當前主從節點信息
        current_master = sentinel.discover_master('my-redis-cluster')
        current_slaves = sentinel.discover_slaves('my-redis-cluster')
        print(f"當前主節點: {current_master}")
        print(f"當前從節點: {current_slaves}")
        
    except RedisError as e:
        print(f"Sentinel 操作失敗: {e}")

# 運行示例
sentinel_operations()

哨兵故障轉移演示

# filename: sentinel_failover.py
import time
import threading

def monitor_sentinel_status():
    """監控哨兵狀態（模擬故障轉移觀察）"""
    sentinel = Sentinel([
        ('localhost', 26379),
        ('localhost', 26380), 
        ('localhost', 26381),
    ], socket_timeout=0.5)
    
    while True:
        try:
            master_addr = sentinel.discover_master('my-redis-cluster')
            slaves = sentinel.discover_slaves('my-redis-cluster')
            print(f"[{time.strftime('%H:%M:%S')}] 主節點: {master_addr}, 從節點: {slaves}")
        except RedisError as e:
            print(f"監控錯誤: {e}")
        
        time.sleep(2)

def sentinel_auto_failover_test():
    """測試哨兵的自動故障轉移"""
    master, slave, sentinel = sentinel_connection()
    
    # 持續寫入數據
    def writer():
        counter = 0
        while True:
            try:
                master.set('failover:test', f'value_{counter}')
                master.incr('failover:counter')
                counter += 1
                time.sleep(1)
            except RedisError as e:
                print(f"寫入失敗: {e}, 等待故障轉移...")
                time.sleep(3)
    
    # 持續讀取數據  
    def reader():
        while True:
            try:
                value = slave.get('failover:test')
                counter = slave.get('failover:counter')
                print(f"讀取: {value}, 計數器: {counter}")
            except RedisError as e:
                print(f"讀取失敗: {e}")
            time.sleep(1)
    
    # 啓動讀寫線程
    threading.Thread(target=writer, daemon=True).start()
    threading.Thread(target=reader, daemon=True).start()
    
    # 運行監控
    monitor_sentinel_status()

# 注意：運行此示例需要真實的哨兵環境
# sentinel_auto_failover_test()

集羣模式（Cluster）

架構概述

  • 數據自動分片到 16384 個槽位（slots）
  • 每個節點負責一部分槽位
  • 客户端直接路由命令到正確的節點
  • 每個主節點都有對應的從節點

Python 連接 Cluster

# filename: cluster_demo.py
def cluster_connection():
    """連接 Redis Cluster"""
    
    # 啓動節點列表（不需要所有節點，能連接一個即可發現整個集羣）
    startup_nodes = [
        {"host": "redis-cluster-node1.example.com", "port": 6379},
        {"host": "redis-cluster-node2.example.com", "port": 6379},
        {"host": "redis-cluster-node3.example.com", "port": 6379},
    ]
    
    try:
        # 方法1: 使用 redis-py-cluster (推薦)
        from redis.cluster import RedisCluster
        
        rc = RedisCluster(
            startup_nodes=startup_nodes,
            password='your_cluster_password',
            decode_responses=True,
            skip_full_coverage_check=True,  # 避免不必要的全集羣檢查
            socket_connect_timeout=5,
            socket_timeout=5,
            retry_on_timeout=True,
            max_connections_per_node=20
        )
        
        return rc
        
    except ImportError:
        # 方法2: 使用 redis-py 內置的集羣支持（基礎功能）
        print("redis-py-cluster 未安裝，使用 redis-py 基礎集羣支持")
        rc = redis.Redis(
            host=startup_nodes[0]['host'],
            port=startup_nodes[0]['port'],
            password='your_cluster_password',
            decode_responses=True
        )
        
        return rc

def cluster_operations():
    """集羣基本操作"""
    try:
        rc = cluster_connection()
        
        # 基本操作 - API 與單機版基本一致
        rc.set('cluster:key1', 'value1')
        rc.set('cluster:key2', 'value2')
        
        value1 = rc.get('cluster:key1')
        value2 = rc.get('cluster:key2')
        print(f"獲取值: key1={value1}, key2={value2}")
        
        # 計數器操作
        rc.incr('cluster:counter')
        counter = rc.get('cluster:counter')
        print(f"計數器: {counter}")
        
        # 獲取集羣信息
        if hasattr(rc, 'cluster_info'):
            info = rc.cluster_info()
            print(f"集羣狀態: {info.get('cluster_state')}")
            print(f"已知節點數: {info.get('cluster_known_nodes')}")
        
        # 獲取鍵所在的槽位和節點
        key_slot = rc.cluster_keyslot('cluster:key1')
        print(f"key1 的槽位: {key_slot}")
        
    except RedisError as e:
        print(f"集羣操作失敗: {e}")

# 運行示例
cluster_operations()

集羣分片與路由

# filename: cluster_sharding.py
def cluster_sharding_demo():
    """演示集羣的數據分片特性"""
    rc = cluster_connection()
    
    # 存儲多個鍵，它們會被自動分配到不同節點
    keys = []
    for i in range(10):
        key = f'sharding:key:{i}'
        rc.set(key, f'value_{i}')
        keys.append(key)
    
    # 驗證鍵分佈在不同的槽位
    for key in keys:
        slot = rc.cluster_keyslot(key)
        # 獲取負責該槽位的節點
        node = rc.nodes_manager.get_node_from_slot(slot)
        print(f"鍵 {key} -> 槽位 {slot} -> 節點 {node}")
    
    # 批量操作的限制：跨槽位的 MSET/MGET 會失敗
    try:
        # 這可能會失敗，因為 keys 可能在不同的槽位
        result = rc.mget(keys)
        print(f"批量獲取成功: {result}")
    except redis.RedisError as e:
        print(f"跨槽位批量操作失敗: {e}")
        # 解決方案：使用 pipeline 或哈希標籤確保鍵在同一個槽位

def cluster_hash_tags():
    """使用哈希標籤確保相關鍵在同一個槽位"""
    rc = cluster_connection()
    
    # 使用 {user123} 作為哈希標籤，確保所有 user123 相關的鍵在同一個槽位
    user_id = "user123"
    
    rc.set(f"user:{{{user_id}}}:profile", "profile_data")
    rc.set(f"user:{{{user_id}}}:session", "session_data") 
    rc.set(f"user:{{{user_id}}}:preferences", "prefs_data")
    
    # 現在可以安全地進行批量操作
    keys = [
        f"user:{{{user_id}}}:profile",
        f"user:{{{user_id}}}:session", 
        f"user:{{{user_id}}}:preferences"
    ]
    
    try:
        values = rc.mget(keys)
        print(f"使用哈希標籤批量獲取: {values}")
    except redis.RedisError as e:
        print(f"哈希標籤批量操作失敗: {e}")

cluster_hash_tags()

性能優化與容量規劃

連接池配置實踐

# filename: production_connection_pools.py
def production_connection_configs():
    """生產環境連接配置示例"""
    
    # Sentinel 連接池配置
    sentinel_pool = redis.sentinel.SentinelConnectionPool(
        'my-redis-cluster',
        redis.sentinel.Sentinel([
            ('sentinel1', 26379),
            ('sentinel2', 26379),
            ('sentinel3', 26379),
        ]),
        password='your_password',
        max_connections=50,
        socket_timeout=5,
        socket_connect_timeout=5,
        retry_on_timeout=True,
        health_check_interval=30
    )
    
    # Cluster 連接池配置
    cluster_pool = RedisCluster(
        startup_nodes=[{"host": "node1", "port": 6379}],
        password='your_password',
        max_connections=100,  # 每個節點的最大連接數
        socket_connect_timeout=5,
        socket_timeout=5,
        retry_on_timeout=True,
        reinitialize_steps=10  # 每10次命令後重新初始化連接
    )

監控與健康檢查

# filename: health_monitoring.py
class RedisHealthMonitor:
    """Redis 健康監控器"""
    
    def __init__(self, client):
        self.client = client
    
    def check_connectivity(self):
        """檢查基本連通性"""
        try:
            return self.client.ping()
        except (ConnectionError, RedisError):
            return False
    
    def get_info(self, section=None):
        """獲取 Redis 信息"""
        try:
            if hasattr(self.client, 'info'):
                return self.client.info(section)
            else:
                # Cluster 的特殊處理
                return self.client.cluster_info()
        except RedisError:
            return None
    
    def monitor_memory_usage(self):
        """監控內存使用情況"""
        info = self.get_info('memory')
        if info:
            return {
                'used_memory': info.get('used_memory_human', 'N/A'),
                'used_memory_peak': info.get('used_memory_peak_human', 'N/A'),
                'memory_fragmentation_ratio': info.get('mem_fragmentation_ratio', 'N/A')
            }
        return None
    
    def check_replication_status(self):
        """檢查複製狀態（適用於主從和集羣）"""
        info = self.get_info('replication')
        if info:
            role = info.get('role')
            if role == 'master':
                return {
                    'role': 'master',
                    'connected_slaves': info.get('connected_slaves', 0)
                }
            elif role == 'slave':
                return {
                    'role': 'slave', 
                    'master_link_status': info.get('master_link_status', 'down'),
                    'master_host': info.get('master_host', 'unknown')
                }
        return None

安全與可靠性

  • 使用安全組/防火牆限制訪問來源
  • Redis 節點部署在私有子網
  • 哨兵和集羣節點間使用專用網絡
  • 配置適當的 save 規則或啓用 AOF
  • 定期測試備份恢復流程
  • 跨可用區部署提高容災能力

案例

雲服務連接示例（AWS ElastiCache）

# filename: aws_elasticache.py
def connect_aws_elasticache():
    """連接 AWS ElastiCache Redis"""
    
    # ElastiCache Redis (Cluster Mode Disabled) - 使用 Sentinel
    if os.getenv('REDIS_MODE') == 'sentinel':
        sentinel = Sentinel([
            ('primary-endpoint', 26379),
            ('secondary-endpoint', 26379),
        ], socket_timeout=1)
        
        client = sentinel.master_for(
            'my-cluster',
            socket_timeout=1,
            password=os.getenv('REDIS_PASSWORD')
        )
    
    # ElastiCache Redis (Cluster Mode Enabled) - 使用 Cluster
    elif os.getenv('REDIS_MODE') == 'cluster':
        # 獲取配置端點
        configuration_endpoint = os.getenv('REDIS_CLUSTER_CONFIG_ENDPOINT')
        host, port = configuration_endpoint.split(':')
        
        client = RedisCluster(
            startup_nodes=[{"host": host, "port": int(port)}],
            password=os.getenv('REDIS_PASSWORD'),
            skip_full_coverage_check=True,
            decode_responses=True
        )
    
    else:
        # 單機模式
        client = redis.Redis(
            host=os.getenv('REDIS_HOST'),
            port=int(os.getenv('REDIS_PORT')),
            password=os.getenv('REDIS_PASSWORD'),
            decode_responses=True
        )
    
    return client

# 生產環境配置管理器
class RedisConnectionManager:
    """生產環境 Redis 連接管理器"""
    
    _clients = {}
    
    @classmethod
    def get_client(cls, service_name='default'):
        """獲取 Redis 客户端（單例模式）"""
        if service_name not in cls._clients:
            if os.getenv('REDIS_CLUSTER_ENABLED') == 'true':
                cls._clients[service_name] = connect_aws_elasticache()
            else:
                # 其他連接邏輯
                pass
        
        # 健康檢查
        try:
            cls._clients[service_name].ping()
        except RedisError:
            # 重新建立連接
            cls._clients[service_name] = connect_aws_elasticache()
        
        return cls._clients[service_name]

小結

從單機到哨兵再到集羣，Redis 提供了完整的解決方案來滿足不同規模應用的高可用和可擴展性需求。哨兵模式為讀寫分離和自動故障轉移提供了優雅的解決方案，而集羣模式則通過數據分片實現了真正的水平擴展。

數據庫
user avatar
0 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.