当前位置： 首页 > news >正文

Redis 键值对操作详解：Python 实现指南

news 2025/8/1 6:23:39

一、环境准备

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, TimeoutError
import timedef safe_operation():attempts = 0max_attempts = 3while attempts < max_attempts:try:# 尝试执行操作return r.set("important:data", "critical_value", ex=30)except (ConnectionError, TimeoutError) as e:attempts += 1print(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. 连接复用：避免频繁创建/关闭连接

六、总结与选择建议

操作选择矩阵

场景推荐操作替代方案
添加小数据SETHSET（对象）
添加大数据HSET/MSET分批次添加
添加临时数据SETEXSET + EXPIRE
删除小数据DELETEUNLINK
删除大数据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 的核心区别，并在不同场景下选择最优的操作策略。

http://www.lryc.cn/news/604220.html

相关文章：

  • 字符串函数安全解析成执行函数
  • 解密数据结构之二叉树
  • Wan2.1
  • “非参数化”大语言模型与RAG的关系？
  • 集成电路学习：什么是Wi-Fi无线保真度
  • 「源力觉醒 创作者计划」_文心大模型 4.5 多模态实测：开源加速 AI 普惠落地
  • LeetCode 283 - 移动零
  • 【面试】软件测试面试题
  • mangoDB面试题及详细答案 117道（026-050）
  • Netty中InternalThreadLocalMap的作用
  • 【C++算法】72.队列+宽搜_二叉树的最大宽度
  • React函数组件的“生活管家“——useEffect Hook详解
  • 【Linux】初识make/makefile
  • sqLite 数据库 (2)：如何复制一张表，事务，聚合函数，分组加过滤，列约束，多表查询，视图，触发器与日志管理，创建索引
  • 【MySQL基础篇】：MySQL表的约束常用类型以及实战示例
  • 15-C语言：第15~16天笔记
  • dubbo应用之3.0新特性(响应式编程)(2)
  • 《剑指offer》-算法篇-位运算
  • window weblogic 解锁
  • Object.freeze() 深度解析：不可变性的实现与实战指南
  • 第4章唯一ID生成器——4.5 美团点评开源方案Leaf
  • JVM易混淆名称
  • 【24】C# 窗体应用WinForm ——日历MonthCalendar属性、方法、实例应用
  • 在依赖关系正确的情况下，执行 mvn install 提示找不到软件包
  • 测试自动化不踩坑：4 策略告别 “为自动化而自动化”
  • DPDK PMD 深度解析：高性能网络的核心引擎
  • 使用LangChain构建法庭预定智能体：结合vLLM部署的Qwen3-32B模型
  • 疯狂星期四文案网第23天运营日记
  • 基于Matlab图像处理的静态雨滴去除与质量评估系统
  • 数学建模算法-day[14]