网站导航
> 技术文档 > Redis 键值对操作详解：Python 实现指南_python redis hset mapping 批量设置字段

Redis 键值对操作详解：Python 实现指南_python redis hset mapping 批量设置字段

网友: 技术文档 2025-08-08 03:49:57

一、环境准备

1. 安装依赖库

pip install redis

2. 连接 Redis 数据库

import redis# 创建 Redis 客户端连接r = redis.Redis( host=\'localhost\', # Redis 服务器地址 port=6379, # Redis 端口 db=0,  # 数据库编号（0~15） password=None, # 密码（若无密码则为 None） decode_responses=True # 自动解码返回值为字符串)# 测试连接try: r.ping() print(\"成功连接到 Redis 服务器\")except redis.ConnectionError: print(\"无法连接 Redis 服务器\")

二、添加键值对操作

1. 添加单个键值对

# 添加/覆盖键值对result = r.set(\"username\", \"john_doe\")print(f\"添加结果: {result}\") # 输出: True# 添加带条件的键值对（仅当键不存在时）result = r.set(\"email\", \"john@example.com\", nx=True)print(f\"条件添加结果: {result}\") # 键不存在时返回 True，存在时返回 None

2. 批量添加键值对

# 批量添加多个键值对data = { \"user:1001:name\": \"Alice\", \"user:1001:age\": \"28\", \"user:1001:city\": \"London\"}result = r.mset(data)print(f\"批量添加结果: {result}\") # 输出: True

3. 添加带过期时间的键值对

# 设置键值对并在60秒后自动过期result = r.setex(\"session_token\", 60, \"a1b2c3d4e5\")print(f\"带过期时间的添加结果: {result}\") # 输出: True# 检查剩余生存时间ttl = r.ttl(\"session_token\")print(f\"剩余生存时间: {ttl}秒\") # 输出: 60

4. 添加大型键值对

# 创建大型数据集large_data = {f\"item_{i}\": f\"value_{i}\" for i in range(10000)}# 添加大型哈希表result = r.hset(\"large:hash\", mapping=large_data)print(f\"添加大型哈希表结果: {result}\") # 输出: 添加的字段数量

三、删除键值对操作

1. DELETE 命令（同步删除）

# 删除单个键result = r.delete(\"username\")print(f\"删除单个键结果: {result}\") # 输出: 1 (成功删除)# 删除多个键keys_to_delete = [\"user:1001:name\", \"user:1001:age\", \"non_existent_key\"]result = r.delete(*keys_to_delete)print(f\"删除多个键结果: {result}\") # 输出: 2 (实际删除的键数)

2. UNLINK 命令（异步删除）

# 异步删除单个大键result = r.unlink(\"large:hash\")print(f\"UNLINK 单个键结果: {result}\") # 输出: 1# 异步删除多个键keys_to_unlink = [\"session_token\", \"temp:data\", \"cache:item\"]result = r.unlink(*keys_to_unlink)print(f\"UNLINK 多个键结果: {result}\") # 输出: 实际删除的键数

3. DELETE vs UNLINK 对比

特性 DELETE 命令 UNLINK 命令 执行方式 同步删除（阻塞操作） 异步删除（非阻塞操作） 适用场景 小型键值对 大型键值对（哈希、列表等） 性能影响 可能阻塞服务器 后台执行，不影响主线程 返回值 删除的键数量 删除的键数量 内存回收 立即回收 延迟回收 Python方法 .delete() .unlink()

4. 删除性能对比测试

import time# 创建大型测试数据large_hash = {f\"key_{i}\": f\"value_{i}\" for i in range(50000)}r.hset(\"test:large:hash\", mapping=large_hash)# 测试 DELETE 性能start = time.time()r.delete(\"test:large:hash\")delete_duration = time.time() - start# 重新创建数据r.hset(\"test:large:hash\", mapping=large_hash)# 测试 UNLINK 性能start = time.time()r.unlink(\"test:large:hash\")unlink_duration = time.time() - startprint(f\"DELETE 耗时: {delete_duration:.4f}秒\")print(f\"UNLINK 耗时: {unlink_duration:.4f}秒\")print(f\"性能差异: DELETE 比 UNLINK 慢 {delete_duration/unlink_duration:.1f}倍\")

四、高级操作技巧

1. 管道操作（批量执行）

# 使用管道批量添加和删除with r.pipeline() as pipe: # 批量添加 pipe.set(\"counter\", 0) pipe.incrby(\"counter\", 100) pipe.set(\"status\", \"active\") # 批量删除 pipe.delete(\"temp:data1\", \"temp:data2\") pipe.unlink(\"large:cache\") # 执行所有命令 results = pipe.execute()print(f\"管道操作结果: {results}\")

2. 哈希表操作

# 添加哈希表r.hset(\"user:1002\", mapping={ \"name\": \"Bob\", \"email\": \"bob@example.com\", \"age\": \"32\"})# 获取哈希表字段name = r.hget(\"user:1002\", \"name\")print(f\"用户名: {name}\")# 删除哈希表字段r.hdel(\"user:1002\", \"age\")# 获取所有字段all_fields = r.hgetall(\"user:1002\")print(f\"用户数据: {all_fields}\")

3. 键存在性检查

# 检查单个键是否存在exists = r.exists(\"username\")print(f\"键存在: {bool(exists)}\") # 输出: True 或 False# 检查多个键是否存在count = r.exists(\"key1\", \"key2\", \"key3\")print(f\"存在的键数量: {count}\")

五、最佳实践与注意事项

1. 键操作选择指南

  • 小型字符串键：DELETE 或 UNLINK 均可
  • 大型数据结构：始终使用 UNLINK
  • 批量删除操作：优先使用 UNLINK + 管道
  • 需要立即释放内存：使用 DELETE
  • 高并发环境：优先使用 UNLINK

2. 内存管理建议

# 监控内存使用情况info = r.info(\"memory\")print(f\"已用内存: {info[\'used_memory_human\']}\")print(f\"待删除对象: {info[\'lazyfree_pending_objects\']}\")

3. 错误处理与重试

from redis.exceptions import ConnectionError, TimeoutErrorimport timedef safe_operation(): attempts = 0 max_attempts = 3 while attempts < max_attempts: try: # 尝试执行操作 return r.set(\"important:data\", \"critical_value\", ex=30) except (ConnectionError, TimeoutError) as e: attempts += 1 print(f\"操作失败 ({attempts}/{max_attempts}): {str(e)}\") time.sleep(2 ** attempts) # 指数退避 print(\"操作失败，达到最大重试次数\") return Falsesafe_operation()

4. 性能优化技巧

  1. 批量操作：使用 MSET 替代多个 SET，使用管道处理批量命令
  2. 键名设计：使用可读的命名空间（如 user:1000:profile）
  3. 过期时间：为临时数据设置 TTL，避免手动删除
  4. 异步删除：大型数据删除始终使用 UNLINK
  5. 连接复用：避免频繁创建/关闭连接

六、总结与选择建议

操作选择矩阵

场景 推荐操作 替代方案 添加小数据 SET HSET（对象） 添加大数据 HSET/MSET 分批次添加 添加临时数据 SETEX SET + EXPIRE 删除小数据 DELETE UNLINK 删除大数据 UNLINK 无 批量操作 管道 + MSET/UNLINK 单独命令

核心要点总结

  1. 添加操作：

    • 使用 set() 添加单个键值对
    • 使用 mset() 批量添加多个键值对
    • 使用 setex() 添加带过期时间的键值对

  2. 删除操作：

    • 优先使用 unlink() 进行删除（尤其大型数据）
    • 仅在需要立即释放内存时使用 delete()
    • 批量删除时结合管道提高效率

  3. 性能关键：

    • UNLINK 比 DELETE 快数百倍（大型数据）
    • 管道操作可减少网络往返时间
    • 合理设置过期时间减少手动删除

  4. 最佳实践：

    • 生产环境默认使用 UNLINK
    • 监控 lazyfree_pending_objects 指标
    • 使用指数退避策略处理连接错误
# 最终推荐操作模式def redis_best_practice(): # 添加数据 r.set(\"app:status\", \"running\", ex=3600) # 带过期时间 r.mset({\"config:theme\": \"dark\", \"config:lang\": \"en\"}) # 删除数据 r.unlink(\"old:cache:data\") # 大型数据 r.unlink(\"temp:session:1\", \"temp:session:2\") # 批量删除 # 使用管道 with r.pipeline() as pipe: pipe.incr(\"counter:requests\") pipe.expire(\"counter:requests\", 86400) pipe.unlink(\"obsolete:key\") pipe.execute()

通过本指南，您应该能够熟练地在 Python 中实现 Redis 的键值对添加和删除操作，理解 DELETE 和 UNLINK 的核心区别，并在不同场景下选择最优的操作策略。

网络标签：

相关问题