### Python操作Redis实现数据增查的完整示例 Redis作为高性能的键值数据库，通过Python操作可以实现高效的数据增删查改操作。下面通过完整的代码示例展示如何使用Python连接Redis并实现数据增加和查询功能。 #### 1. 环境准备与连接配置 首先需要安装redis-py库，这是Python操作Redis的标准库： ```bash pip install redis ``` 建立Redis连接的两种方式： ```python import redis # 方式1：直接连接 r = redis.Redis( host='localhost', # Redis服务器地址 port=6379, # Redis端口 db=0, # 数据库编号 password=None, # 密码（如果有） decode_responses=True # 自动解码返回字符串 ) # 方式2：连接池（推荐用于生产环境） pool = redis.ConnectionPool( host='localhost', port=6379, db=0, max_connections=10, # 最大连接数 decode_responses=True ) r = redis.Redis(connection_pool=pool) ``` 连接池方式能够有效管理连接资源，避免频繁创建和关闭连接的开销 [ref_2]。 #### 2. 字符串数据的增查操作 字符串是Redis最基本的数据类型，适用于存储简单的键值对： ```python def string_operations(): """字符串类型数据的增加和查询操作""" # 增加数据 - 设置键值对 r.set('user:1001:name', '张三') # 设置用户姓名 r.set('user:1001:age', '25') # 设置用户年龄 r.set('user:1001:email', 'zhangsan@example.com') # 带过期时间的设置（30分钟过期） r.setex('user:1001:session', 1800, 'active') # 查询数据 name = r.get('user:1001:name') # 获取姓名 age = r.get('user:1001:age') # 获取年龄 email = r.get('user:1001:email') # 获取邮箱 session = r.get('user:1001:session') # 获取会话状态 print(f"用户姓名: {name}") print(f"用户年龄: {age}") print(f"用户邮箱: {email}") print(f"会话状态: {session}") # 检查键是否存在 if r.exists('user:1001:name'): print("用户姓名键存在") # 获取多个键的值 keys = ['user:1001:name', 'user:1001:age', 'user:1001:email'] values = r.mget(keys) print(f"批量查询结果: {values}") # 执行字符串操作示例 string_operations() ``` #### 3. 哈希数据的增查操作 哈希类型适合存储对象，可以将多个字段存储在一个键中： ```python def hash_operations(): """哈希类型数据的增加和查询操作""" # 增加数据 - 设置用户信息的哈希字段 user_data = { 'name': '李四', 'age': '30', 'city': '北京', 'occupation': '工程师' } # 一次性设置多个字段 r.hset('user:1002', mapping=user_data) # 单独添加字段 r.hset('user:1002', 'salary', '15000') # 查询数据 # 获取单个字段 name = r.hget('user:1002', 'name') age = r.hget('user:1002', 'age') # 获取所有字段 all_fields = r.hgetall('user:1002') # 获取多个指定字段 specific_fields = r.hmget('user:1002', ['name', 'city', 'occupation']) # 检查字段是否存在 if r.hexists('user:1002', 'city'): city = r.hget('user:1002', 'city') print(f"用户所在城市: {city}") print(f"用户姓名: {name}") print(f"用户年龄: {age}") print(f"所有字段: {all_fields}") print(f"指定字段: {specific_fields}") # 获取所有字段名 fields = r.hkeys('user:1002') print(f"所有字段名: {fields}") # 获取所有字段值 values = r.hvals('user:1002') print(f"所有字段值: {values}") # 执行哈希操作示例 hash_operations() ``` #### 4. 列表数据的增查操作 列表类型支持有序的元素集合，适用于消息队列等场景： ```python def list_operations(): """列表类型数据的增加和查询操作""" # 增加数据 - 向列表中添加元素 key = 'user:1003:messages' # 从左侧插入（LPUSH） r.lpush(key, '消息1', '消息2', '消息3') # 从右侧插入（RPUSH） r.rpush(key, '最新消息') # 在指定元素前后插入 r.linsert(key, 'BEFORE', '消息2', '插入的消息') # 查询数据 # 获取列表长度 length = r.llen(key) print(f"消息列表长度: {length}") # 获取指定范围的元素 all_messages = r.lrange(key, 0, -1) # 获取所有元素 first_three = r.lrange(key, 0, 2) # 获取前3个元素 # 获取指定位置的元素 first_msg = r.lindex(key, 0) # 获取第一个元素 last_msg = r.lindex(key, -1) # 获取最后一个元素 print(f"所有消息: {all_messages}") print(f"前三个消息: {first_three}") print(f"第一条消息: {first_msg}") print(f"最后一条消息: {last_msg}") # 执行列表操作示例 list_operations() ``` #### 5. 集合数据的增查操作 集合类型存储不重复的无序元素，适用于标签、好友关系等场景： ```python def set_operations(): """集合类型数据的增加和查询操作""" # 增加数据 - 向集合中添加元素 tags_key = 'article:1001:tags' # 添加标签 r.sadd(tags_key, 'Python', 'Redis', '数据库', '编程') r.sadd(tags_key, 'Python') # 重复添加会被忽略 # 查询数据 # 获取所有成员 all_tags = r.smembers(tags_key) # 获取集合大小 tag_count = r.scard(tags_key) # 检查元素是否存在 has_python = r.sismember(tags_key, 'Python') has_java = r.sismember(tags_key, 'Java') # 随机获取元素 random_tag = r.srandmember(tags_key) print(f"所有标签: {all_tags}") print(f"标签数量: {tag_count}") print(f"是否包含Python: {has_python}") print(f"是否包含Java: {has_java}") print(f"随机标签: {random_tag}") # 执行集合操作示例 set_operations() ``` #### 6. 完整综合示例 下面是一个完整的用户管理系统示例，展示多种数据类型的综合应用： ```python def comprehensive_user_management(): """综合用户管理系统示例""" # 清空测试数据 r.flushdb() print("=== 用户管理系统演示 ===") # 1. 存储用户基本信息（字符串） user_id = "1001" base_key = f"user:{user_id}" # 设置用户基本信息 user_info = { f"{base_key}:name": "王五", f"{base_key}:age": "28", f"{base_key}:email": "wangwu@example.com", f"{base_key}:status": "active" } # 批量设置 r.mset(user_info) # 2. 存储用户详细信息（哈希） user_detail = { 'phone': '13800138000', 'address': '上海市浦东新区', 'department': '技术部', 'position': '高级工程师' } r.hset(f"{base_key}:detail", mapping=user_detail) # 3. 存储用户技能标签（集合） skills = ['Python', 'Redis', 'MySQL', 'Docker', 'Linux'] r.sadd(f"{base_key}:skills", *skills) # 4. 存储用户操作日志（列表） logs = [ '2024-01-15 10:00:00 用户登录', '2024-01-15 10:30:00 查看个人资料', '2024-01-15 11:00:00 修改密码' ] r.rpush(f"{base_key}:logs", *logs) print("\n=== 查询用户信息 ===") # 查询基本信息 base_info = r.mget([f"{base_key}:name", f"{base_key}:age", f"{base_key}:email"]) print(f"基本信息 - 姓名: {base_info[0]}, 年龄: {base_info[1]}, 邮箱: {base_info[2]}") # 查询详细信息 detail_info = r.hgetall(f"{base_key}:detail") print(f"详细信息: {detail_info}") # 查询技能标签 user_skills = r.smembers(f"{base_key}:skills") print(f"技能标签: {user_skills}") # 查询最近的操作日志 recent_logs = r.lrange(f"{base_key}:logs", -3, -1) # 最近3条日志 print(f"最近操作: {recent_logs}") # 统计信息 total_skills = r.scard(f"{base_key}:skills") total_logs = r.llen(f"{base_key}:logs") print(f"统计 - 技能数: {total_skills}, 日志数: {total_logs}") # 执行综合示例 comprehensive_user_management() ``` #### 7. 错误处理与最佳实践 在实际应用中，需要添加适当的错误处理： ```python def safe_redis_operations(): """带错误处理的Redis操作""" try: # 测试连接 if r.ping(): print("Redis连接正常") # 执行数据操作 r.set('test_key', 'test_value') value = r.get('test_key') print(f"测试键值: {value}") except redis.ConnectionError as e: print(f"Redis连接错误: {e}") except redis.TimeoutError as e: print(f"Redis操作超时: {e}") except Exception as e: print(f"其他错误: {e}") finally: # 清理测试数据 r.delete('test_key') # 执行安全操作 safe_redis_operations() ``` #### 关键技术要点总结 | 操作类型 | 主要方法 | 适用场景 | 特点 | |---------|---------|---------|------| | 字符串操作 | set, get, mset, mget | 简单键值存储、计数器 | 简单高效，支持批量操作 | | 哈希操作 | hset, hget, hgetall | 对象存储、用户属性 | 结构化数据，字段级操作 | | 列表操作 | lpush, rpush, lrange | 消息队列、时间线 | 有序集合，支持范围查询 | | 集合操作 | sadd, smembers, sismember | 标签系统、好友关系 | 去重存储，支持集合运算 | 通过上述示例，可以看到Python操作Redis进行数据增加和查询的完整流程。在实际项目中，建议使用连接池管理连接，合理选择数据类型，并添加适当的错误处理机制 [ref_1][ref_2]。Redis的高性能特性使其特别适合缓存、会话存储、消息队列等场景，结合Python的简洁语法，能够快速构建高效的应用程序。