我要评论一、环境准备
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() - start
print(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 time
def 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 false
safe_operation()
4. 性能优化技巧
- 批量操作：使用 mset 替代多个 set，使用管道处理批量命令
- 键名设计：使用可读的命名空间（如
user:1000:profile） - 过期时间：为临时数据设置 ttl，避免手动删除
- 异步删除：大型数据删除始终使用 unlink
- 连接复用：避免频繁创建/关闭连接
六、总结与选择建议
操作选择矩阵
| 场景 | 推荐操作 | 替代方案 |
|---|---|---|
| 添加小数据 | set | hset（对象） |
| 添加大数据 | hset/mset | 分批次添加 |
| 添加临时数据 | setex | set + expire |
| 删除小数据 | delete | unlink |
| 删除大数据 | unlink | 无 |
| 批量操作 | 管道 + mset/unlink | 单独命令 |
核心要点总结
添加操作：
- 使用
set()添加单个键值对 - 使用
mset()批量添加多个键值对 - 使用
setex()添加带过期时间的键值对
删除操作：
- 优先使用
unlink()进行删除（尤其大型数据） - 仅在需要立即释放内存时使用
delete() - 批量删除时结合管道提高效率
性能关键：
- unlink 比 delete 快数百倍（大型数据）
- 管道操作可减少网络往返时间
- 合理设置过期时间减少手动删除
最佳实践：
- 生产环境默认使用 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 的核心区别，并在不同场景下选择最优的操作策略。
以上就是在python实现redis键值对添加和删除的操作指南的详细内容，更多关于python redis键值对添加和删除的资料请关注代码网其它相关文章！
发表评论