### Redis 连接与操作的完整 Python 类 以下是基于 `redis-py` 库实现的一个功能齐全的 Python 类，用于连接和操作 Redis。此类封装了常见的 Redis 操作，包括设置键值对、获取键值对、删除键、订阅发布等功能。 ```python import redis from typing import Any, List, Dict class RedisClient: def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0, password: str = None): """ 初始化 Redis 客户端。 参数: host (str): Redis 主机地址，默认为 localhost。 port (int): Redis 端口号，默认为 6379。 db (int): 使用的数据库编号，默认为 0。 password (str): 密码，如果不需要则设为 None。 """ self.connection = redis.StrictRedis(host=host, port=port, db=db, password=password, decode_responses=True) def set_key(self, key: str, value: Any, ex: int = None) -> bool: """ 设置一个键值对。 参数: key (str): 键名。 value (Any): 值。 ex (int): 可选参数，过期时间（秒）。 返回: 成功返回 True，失败返回 False。 """ try: result = self.connection.set(key, value, ex=ex) return result is not None except Exception as e: print(f"Error setting key: {e}") return False def get_key(self, key: str) -> Any: """ 获取指定键的值。 参数: key (str): 键名。 返回: 如果存在，则返回对应的值；否则返回 None。 """ try: return self.connection.get(key) except Exception as e: print(f"Error getting key: {e}") return None def delete_key(self, key: str) -> int: """ 删除指定键。 参数: key (str): 键名。 返回: 被删除键的数量。 """ try: return self.connection.delete(key) except Exception as e: print(f"Error deleting key: {e}") return 0 def publish_message(self, channel: str, message: str) -> int: """ 发布消息到指定频道。 参数: channel (str): 频道名称。 message (str): 消息内容。 返回: 接收到消息的订阅者数量。 """ try: return self.connection.publish(channel, message) except Exception as e: print(f"Error publishing message: {e}") return 0 def subscribe_channel(self, channels: List[str]) -> redis.client.PubSub: """ 订阅指定频道。 参数: channels (List[str]): 频道列表。 返回: PubSub 对象。 """ pubsub = self.connection.pubsub() pubsub.subscribe(channels) return pubsub def hset_hash(self, name: str, key: str, value: Any) -> int: """ 向哈希表中存储键值对。 参数: name (str): 哈希表名称。 key (str): 键名。 value (Any): 值。 返回: 执行成功返回 1，已存在更新返回 0。 """ try: return self.connection.hset(name, key, value) except Exception as e: print(f"Error setting hash: {e}") return 0 def hget_hash(self, name: str, key: str) -> Any: """ 从哈希表中获取指定键的值。 参数: name (str): 哈希表名称。 key (str): 键名。 返回: 如果存在，则返回对应的值；否则返回 None。 """ try: return self.connection.hget(name, key) except Exception as e: print(f"Error getting hash: {e}") return None def hmset_hash(self, name: str, mapping: Dict[str, Any]) -> bool: """ 向哈希表中批量存储键值对。 参数: name (str): 哈希表名称。 mapping (Dict[str, Any]): 键值对字典。 返回: 成功返回 True，失败返回 False。 """ try: return self.connection.hmset(name, mapping) except Exception as e: print(f"Error batch setting hash: {e}") return False def close_connection(self): """ 关闭 Redis 连接。 """ try: self.connection.close() except Exception as e: print(f"Error closing connection: {e}") # 示例用法 if __name__ == "__main__": client = RedisClient(host='127.0.0.1', port=6379, db=0, password=None) # 设置键值对 client.set_key('test_key', 'value_1', ex=10) # 设置带过期时间 # 获取键值对 value = client.get_key('test_key') print(f'Value of test_key: {value}') # 删除键 deleted_count = client.delete_key('test_key') print(f'Deleted keys count: {deleted_count}') # 发布消息 published_count = client.publish_message('channel_1', 'Hello Channel!') print(f'Message received by subscribers: {published_count}') # 订阅频道 subscriber = client.subscribe_channel(['channel_1']) for msg in subscriber.listen(): if msg['type'] == 'message': print(f'Received Message: {msg["data"]}') break # 哈希操作 client.hset_hash('hash_name', 'field_1', 'value_1') # 单条插入 field_value = client.hget_hash('hash_name', 'field_1') print(f'Hash Field Value: {field_value}') client.hmset_hash('hash_name', {'field_2': 'value_2', 'field_3': 'value_3'}) # 批量插入 # 关闭连接 client.close_connection() ``` 以上代码实现了 Redis 的基本操作，并提供了异常捕获机制以提高稳定性[^1]。