网站导航
> 技术文档 > Python 操作 Redis 全指南：从入门到实战_python redis

Python 操作 Redis 全指南：从入门到实战_python redis

网友: 技术文档 2025-07-30 17:41:37

Redis 作为一款高性能的内存数据库，在缓存、会话存储、实时分析等场景中应用广泛。本文将详细介绍如何在 Python 中操作 Redis，涵盖环境搭建、基础数据类型操作、高级功能及最佳实践，适合 Redis 初学者和 Python 开发者参考。

一、环境准备

1. 安装 Redis 服务器

  • Windows：从Redis 官方仓库下载安装包，或使用 WSL2 安装 Linux 版本
  • Linux：通过包管理器安装sudo apt install redis-server（Ubuntu）或yum install redis（CentOS）
  • 验证安装：启动服务后执行redis-cli ping，返回PONG表示运行正常

2. 安装 Python 客户端

使用官方推荐的redis-py库：



pip install redis


# 如需操作Redis集群，额外安装


pip install redis-py-cluster


二、建立连接


1. 基础连接方式




import redis


# 基本连接配置


r = redis.Redis(


host=\'localhost\', # 服务器地址，默认localhost


port=6379, # 端口号，默认6379


password=\'your_pwd\', # 认证密码，无密码可省略


db=0, # 数据库编号（0-15），默认0


decode_responses=True # 自动将返回值转为字符串（默认返回bytes）


)


# 测试连接


try:


if r.ping():


print(\"✅ Redis连接成功\")


except redis.ConnectionError:


print(\"❌ Redis连接失败\")


2. 连接池（推荐）


连接池能复用连接，减少频繁创建连接的开销：




# 创建连接池


pool = redis.ConnectionPool(


host=\'localhost\',


port=6379,


max_connections=20, # 最大连接数


decode_responses=True


)


# 从连接池获取连接


r = redis.Redis(connection_pool=pool)


三、核心数据类型操作


1. String（字符串）


字符串是 Redis 最基础的数据类型，可存储文本、数字等，最大容量 512MB。




# 设置值


r.set(\'username\', \'zhangsan\') # 基本设置


r.setex(\'verify_code\', 300, \'654321\') # 设置5分钟过期的验证码


r.mset({\'age\': \'25\', \'gender\': \'male\'}) # 批量设置


# 获取值


print(r.get(\'username\')) # 输出: zhangsan


print(r.mget([\'age\', \'gender\'])) # 批量获取: [\'25\', \'male\']


# 数值操作


r.set(\'views\', 100)


r.incr(\'views\') # 自增1 → 101


r.incrby(\'views\', 5) # 增加5 → 106


r.decr(\'views\', 3) # 减少3 → 103


2. Hash（哈希）


哈希适合存储对象型数据（如用户信息），由字段 - 值对组成，类似 Python 字典。




# 设置字段


r.hset(\'user:1001\', \'name\', \'李四\') # 单个字段设置


r.hset(\'user:1001\', mapping={ # 批量设置


\'age\': 30,


\'city\': \'深圳\',


\'email\': \'lisi@example.com\'


})


# 获取数据


print(r.hget(\'user:1001\', \'name\')) # 获取单个字段 → 李四


print(r.hgetall(\'user:1001\')) # 获取所有字段 → 完整字典


print(r.hkeys(\'user:1001\')) # 获取所有字段名 → [\'name\',\'age\',...]


print(r.hvals(\'user:1001\')) # 获取所有值 → [\'李四\',\'30\',...]


# 其他操作


r.hdel(\'user:1001\', \'email\') # 删除字段


print(r.hexists(\'user:1001\', \'city\')) # 判断字段是否存在 → True


3. List（列表）


Redis 列表是有序可重复的元素集合，底层实现为双向链表，适合实现队列、栈等结构。




# 添加元素


r.lpush(\'fruit\', \'apple\', \'banana\') # 左侧插入 → [banana, apple]


r.rpush(\'fruit\', \'orange\') # 右侧插入 → [banana, apple, orange]


# 获取元素


print(r.lrange(\'fruit\', 0, -1)) # 获取所有元素 → 完整列表


print(r.lindex(\'fruit\', 1)) # 获取索引1的元素 → apple


print(r.llen(\'fruit\')) # 获取长度 → 3


# 移除元素


print(r.lpop(\'fruit\')) # 左侧弹出 → banana


print(r.rpop(\'fruit\')) # 右侧弹出 → orange


4. Set（集合）


无序且唯一的元素集合，支持交集、并集等集合运算，适合标签、好友关系等场景。




# 添加元素


r.sadd(\'tags\', \'python\', \'redis\', \'database\') # 添加多个元素


# 获取数据


print(r.smembers(\'tags\')) # 查看所有元素 → 无序集合


print(r.scard(\'tags\')) # 元素数量 → 3


print(r.sismember(\'tags\', \'java\')) # 判断是否存在 → False


# 集合运算


r.sadd(\'tags2\', \'redis\', \'mysql\', \'python\')


print(r.sinter(\'tags\', \'tags2\')) # 交集 → {\'python\', \'redis\'}


print(r.sunion(\'tags\', \'tags2\')) # 并集 → {\'python\',\'redis\',\'database\',\'mysql\'}


5. Sorted Set（有序集合）


在集合基础上为每个元素分配分数（score），支持按分数排序和范围查询，适合排行榜、计分系统。




# 添加元素（分数, 成员）


r.zadd(\'ranking\', {


\'张三\': 95,


\'李四\': 88,


\'王五\': 92


})


# 范围查询


# 按分数升序取前2名


print(r.zrange(\'ranking\', 0, 1, withscores=True)) # [(李四,88.0), (王五,92.0)]


# 按分数降序取所有


print(r.zrevrange(\'ranking\', 0, -1, withscores=True)) # [(张三,95.0), (王五,92.0), (李四,88.0)]


# 分数操作


print(r.zscore(\'ranking\', \'张三\')) # 获取分数 → 95.0


r.zincrby(\'ranking\', 3, \'李四\') # 增加分数 → 91.0


四、高级功能


1. 事务处理


使用pipeline实现事务，确保多个命令的原子性执行：




try:


with r.pipeline() as pipe:


# 开启事务


pipe.multi()


# 批量添加命令


pipe.set(\'key1\', \'val1\')


pipe.hset(\'user:1002\', \'name\', \'wangwu\')


pipe.incr(\'counter\')


# 执行事务


results = pipe.execute()


print(f\"事务执行成功: {results}\")


except redis.RedisError as e:


print(f\"事务执行失败: {e}\")


2. 发布订阅机制


实现简单的消息队列功能：




# 发布者


def publisher():


r.publish(\'news\', \'Redis发布订阅示例\')


r.publish(\'news\', \'Python操作Redis指南\')


# 订阅者


def subscriber():


pubsub = r.pubsub()


pubsub.subscribe(\'news\') # 订阅频道


for message in pubsub.listen():


if message[\'type\'] == \'message\':


print(f\"收到消息: {message[\'data\']}\")


# 可添加退出条件


if message[\'data\'] == \'exit\':


break


# 实际使用时需多线程运行


import threading


threading.Thread(target=subscriber, daemon=True).start()


publisher()


3. 过期时间管理




# 设置过期时间


r.set(\'temp_data\', \'临时数据\')


r.expire(\'temp_data\', 60) # 60秒后过期


# 查看剩余时间（-1：永不过期，-2：已过期）


print(r.ttl(\'temp_data\')) # 剩余秒数


# 清除过期时间


r.persist(\'temp_data\')


五、异常处理与最佳实践


1. 常见异常处理




try:


# 执行Redis操作


r.set(\'key\', \'value\')


except redis.ConnectionError:


print(\"连接异常：请检查Redis服务是否运行\")


except redis.AuthenticationError:


print(\"认证失败：密码错误\")


except redis.ResponseError as e:


print(f\"命令执行错误：{e}\")


except Exception as e:


print(f\"其他错误：{e}\")


2. 性能优化建议


    

  • 使用连接池：减少 TCP 连接开销，推荐设置max_connections=10-50
    • 



    

  • 批量操作：优先使用mset、mget等批量命令，减少网络往返
    • 



    

  • 合理设置数据结构：例如存储用户信息用 Hash 而非多个 String
    • 



    

  • 避免大键：单个键值不宜超过 100MB，大列表建议分片存储
    • 



    

  • 定期清理过期数据：利用EXPIRE自动淘汰无用数据
    • 



六、总结


本文介绍了 Python 操作 Redis 的核心方法，从基础连接到高级功能，覆盖了实际开发中的常见场景。Redis 作为高性能的内存数据库，结合 Python 的简洁语法，能极大提升数据处理效率。建议在实际项目中根据业务场景选择合适的数据结构，并遵循最佳实践确保系统稳定性。


后续可深入学习 Redis 的分布式锁、Lua 脚本、集群部署等高级主题，进一步发挥 Redis 的强大功能.



				
				               	

                			

				                

                	

                    	网络标签：                      

                

				
             

		

    

			
    
		

		
	

	     
	

    	

		

    
相关问题