Python 读写 Redis 缓存数据库：写给 Python 初学者的入门案例

Python 读写 Redis 缓存数据库写给 Python 初学者的入门案例很多学习 Python 的小伙伴在学完文件读写、SQLite、MySQL 之后都会接触到一个新的数据库Redis。Redis 和传统关系型数据库不太一样。MySQL、SQLite 更擅长长期保存结构化数据而 Redis 更常用于缓存、计数器、排行榜、验证码、分布式锁、消息队列等高性能场景。这篇文章会先简单介绍 Redis 是什么然后通过 Python 代码演示如何连接 Redis并完成常见的读写操作。一、Redis 是什么Redis 全称是 Remote Dictionary Server可以理解为一个高性能的内存型 Key-Value 数据库。它的核心特点是数据主要存放在内存中读写速度非常快使用 Key-Value 形式保存数据支持丰富的数据结构不只是简单字符串支持设置过期时间适合做缓存支持持久化可以把内存数据保存到磁盘常用于缓存、计数、排行榜、会话、限流等场景我们可以把 Redis 想象成一个超快的字典cache{user:1:name:Alice,article:100:view_count:358,login:code:13800138000:829316}只不过这个字典不是保存在 Python 程序里而是保存在 Redis 服务中。多个程序、多个接口、多个服务器都可以一起访问它。二、Redis 和 MySQL、SQLite 有什么区别简单对比一下对比项RedisMySQL / SQLite存储方式主要在内存中主要在磁盘中数据模型Key-Value 和多种数据结构表、行、列读写速度非常快相对较慢常见用途缓存、计数、排行榜、验证码业务数据持久化查询方式通过 key 访问SQL 查询是否适合复杂查询不适合适合所以 Redis 通常不是用来完全替代 MySQL 或 SQLite 的而是和它们配合使用。例如用户资料保存在 MySQL 中热门用户资料缓存到 Redis 中下次查询时先查 RedisRedis 没有命中再查 MySQL查到后再写回 Redis这个模式通常叫缓存旁路也叫 Cache Aside。三、准备 Redis 环境如果你本机已经安装 Redis可以直接启动 Redis 服务。如果没有安装也可以使用 Docker 快速启动dockerrun--nameredis-demo-p6379:6379-dredis测试 Redis 是否可用redis-cliping如果返回PONG说明 Redis 服务已经启动成功。四、安装 Python Redis 客户端Python 操作 Redis 常用的库是redis。安装命令pipinstallredis安装完成后可以在 Python 中导入importredis五、连接 Redis新建一个文件redis_demo.pyimportredis rredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)print(r.ping())运行python redis_demo.py如果输出True说明 Python 已经成功连接到 Redis。这里的几个参数含义如下参数说明hostRedis 服务器地址portRedis 端口默认 6379dbRedis 数据库编号默认 0decode_responses是否把返回值自动解码成字符串如果不设置decode_responsesTrueRedis 返回的字符串可能是字节类型bAlice对初学者来说建议先加上decode_responsesTrue这样输出更直观。六、字符串 String最基础的缓存读写Redis 最简单的数据类型是 String。可以用它保存用户名、验证码、Token、简单配置、缓存结果等。importredis rredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)# 写入字符串r.set(user:1:name,Alice)# 读取字符串namer.get(user:1:name)print(name)输出Alice设置过期时间缓存通常不应该永久存在所以 Redis 支持给 key 设置过期时间。例如保存短信验证码60 秒后自动删除r.set(login:code:13800138000,829316,ex60)coder.get(login:code:13800138000)print(code)其中ex60表示 60 秒后过期。也可以先写入再单独设置过期时间r.set(temp:data,hello)r.expire(temp:data,30)查看剩余过期时间ttlr.ttl(temp:data)print(ttl)七、数字自增计数器案例Redis 很适合做计数器例如文章阅读量、点赞数、接口访问次数。article_keyarticle:100:view_countr.set(article_key,0)r.incr(article_key)r.incr(article_key)r.incr(article_key)countr.get(article_key)print(count)输出3也可以一次增加指定数量r.incrby(article_key,10)print(r.get(article_key))对应减少r.decr(article_key)r.decrby(article_key,5)计数器是 Redis 非常常见的使用场景因为这类操作简单、高频而且对速度要求高。八、Hash保存对象信息如果要保存一个用户对象用多个 String 也可以user:1:name user:1:age user:1:city但更推荐使用 Hashr.hset(user:1,mapping{name:Alice,age:20,city:Shanghai})namer.hget(user:1,name)print(name)userr.hgetall(user:1)print(user)输出Alice{name:Alice,age:20,city:Shanghai}修改 Hash 中的某个字段r.hset(user:1,city,Beijing)删除某个字段r.hdel(user:1,age)判断字段是否存在existsr.hexists(user:1,name)print(exists)Hash 非常适合保存用户资料、商品摘要、文章基础信息等对象型数据。九、List保存列表数据Redis 的 List 可以理解为一个有序列表。常见用途最新消息列表简单队列最近浏览记录日志缓冲示例keyuser:1:recent_articlesr.delete(key)r.lpush(key,article:100)r.lpush(key,article:101)r.lpush(key,article:102)articlesr.lrange(key,0,-1)print(articles)输出[article:102,article:101,article:100]lpush是从左侧插入所以最后插入的元素排在最前面。如果想从右侧插入可以使用r.rpush(key,article:103)从列表中弹出一个元素itemr.lpop(key)print(item)限制列表长度只保留最近 3 条r.ltrim(key,0,2)这个操作很适合做“最近浏览记录”。十、Set保存不重复集合Set 是无序且不重复的集合。常见用途用户标签点赞用户集合已签到用户集合去重统计示例记录给文章点赞的用户。keyarticle:100:liked_usersr.delete(key)r.sadd(key,user:1)r.sadd(key,user:2)r.sadd(key,user:1)usersr.smembers(key)print(users)is_likedr.sismember(key,user:1)print(is_liked)输出{user:1,user:2}True虽然添加了两次user:1但 Set 会自动去重。统计集合元素数量countr.scard(key)print(count)取消点赞r.srem(key,user:1)十一、Sorted Set排行榜案例Sorted Set 是有序集合每个元素都有一个分数。它非常适合做排行榜。例如游戏积分排行榜keygame:rankr.delete(key)r.zadd(key,{Alice:1500,Bob:1800,Cindy:1700,David:1200})top3r.zrevrange(key,0,2,withscoresTrue)print(top3)输出[(Bob,1800.0),(Cindy,1700.0),(Alice,1500.0)]给用户增加积分r.zincrby(key,200,Alice)查询 Alice 的排名rankr.zrevrank(key,Alice)print(rank)注意Redis 的排名从 0 开始。如果返回 0表示第一名。十二、保存 JSON 数据有时我们希望把一个 Python 字典直接缓存起来。可以使用json.dumps()转成字符串保存读取时再用json.loads()转回来。importjsonimportredis rredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)user{id:1,name:Alice,age:20,roles:[admin,editor]}r.set(cache:user:1,json.dumps(user,ensure_asciiFalse),ex300)cached_textr.get(cache:user:1)cached_userjson.loads(cached_text)print(cached_user[name])print(cached_user[roles])这种方式很适合缓存接口返回结果。不过要注意Redis 本身保存的是字符串。如果你要频繁修改对象中的某个字段Hash 可能比 JSON 字符串更方便。十三、完整案例使用 Redis 缓存用户资料下面做一个更贴近实际项目的案例。假设我们有一个“慢查询函数”它模拟从数据库查询用户资料。为了提高速度我们先查 Redis如果 Redis 中没有再查数据库并把结果写入 Redis。importjsonimporttimeimportredis rredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)defquery_user_from_database(user_id):模拟数据库查询。print(正在查询数据库...)time.sleep(2)fake_database{1:{id:1,name:Alice,age:20},2:{id:2,name:Bob,age:22},3:{id:3,name:Cindy,age:19},}returnfake_database.get(user_id)defget_user(user_id):优先从 Redis 缓存读取用户资料。cache_keyfcache:user:{user_id}cached_userr.get(cache_key)ifcached_user:print(命中 Redis 缓存)returnjson.loads(cached_user)userquery_user_from_database(user_id)ifuserisNone:returnNoner.set(cache_key,json.dumps(user,ensure_asciiFalse),ex60)print(已写入 Redis 缓存)returnuserif__name____main__:print(get_user(1))print(get_user(1))第一次运行时输出类似正在查询数据库... 已写入 Redis 缓存{id:1,name:Alice,age:20}命中 Redis 缓存{id:1,name:Alice,age:20}第一次查询 Redis 没有数据所以会访问“数据库”并等待 2 秒。第二次查询时Redis 中已经有缓存所以直接返回速度明显更快。这就是 Redis 最典型的缓存使用方式。十四、封装一个简单的 Redis 工具类在项目中我们通常不会把 Redis 操作散落在各个地方。可以先封装一个简单工具类importjsonimportredisclassRedisCache:def__init__(self,hostlocalhost,port6379,db0):self.clientredis.Redis(hosthost,portport,dbdb,decode_responsesTrue)defset_json(self,key,value,expire_seconds300):textjson.dumps(value,ensure_asciiFalse)self.client.set(key,text,exexpire_seconds)defget_json(self,key):textself.client.get(key)iftextisNone:returnNonereturnjson.loads(text)defdelete(self,key):returnself.client.delete(key)defexists(self,key):returnself.client.exists(key)1cacheRedisCache()cache.set_json(user:1,{id:1,name:Alice},expire_seconds60)usercache.get_json(user:1)print(user)print(cache.exists(user:1))cache.delete(user:1)这样后续业务代码会更清晰。十五、连接池更适合项目使用在 Web 项目中不建议每次请求都重新创建 Redis 连接。可以使用连接池importredis poolredis.ConnectionPool(hostlocalhost,port6379,db0,decode_responsesTrue,max_connections20)rredis.Redis(connection_poolpool)r.set(site:name,Python Redis Demo)print(r.get(site:name))连接池可以复用连接减少频繁创建连接带来的开销。如果你以后使用 Flask、FastAPI、Django也可以在应用启动时创建 Redis 客户端然后在接口中复用它。十六、删除 key 和清理数据库删除一个 keyr.delete(user:1)判断 key 是否存在existsr.exists(user:1)print(exists)查看匹配的 keykeysr.keys(user:*)print(keys)清空当前数据库r.flushdb()清空所有数据库r.flushall()注意flushdb()和flushall()都是危险操作真实项目中不要随便执行。十七、初学者常见问题1. 连接失败常见错误redis.exceptions.ConnectionError可能原因Redis 服务没有启动host 或 port 写错了Docker 容器没有映射 6379 端口Redis 设置了密码但代码中没有配置如果 Redis 设置了密码需要这样连接rredis.Redis(hostlocalhost,port6379,passwordyour_password,decode_responsesTrue)2. 读取结果是字节类型如果看到bAlice可以在连接时加上decode_responsesTrue3. JSON 读取报错如果执行json.loads(text)报错通常说明 Redis 中保存的内容不是合法 JSON 字符串。建议保存和读取都封装成统一方法避免同一个 key 有时存普通字符串有时存 JSON。4. 缓存数据不是最新的缓存会带来一个新问题数据库更新了但 Redis 里还是旧数据。常见处理方式更新数据库后删除缓存设置较短的过期时间对一致性要求很高的数据不要随便缓存例如defupdate_user_name(user_id,new_name):update_database_user_name(user_id,new_name)r.delete(fcache:user:{user_id})这样下次查询时会重新从数据库读取最新数据再写入 Redis。十八、Redis key 命名建议Redis 的 key 最好有清晰的命名规则。推荐格式业务名:对象名:对象ID:字段名例如user:1:profile user:1:recent_articles article:100:view_count article:100:liked_users login:code:13800138000 game:rank好的 key 命名可以让排查问题更容易。不推荐随意命名a test data1 abc项目变大后这类 key 很难维护。十九、什么时候适合使用 Redis适合使用 Redis 的场景高频读取的数据可以设置过期时间的数据允许短时间不一致的数据计数器、排行榜、点赞集合验证码、Token、Session接口限流、简单队列不适合使用 Redis 的场景复杂 SQL 查询强事务业务长期保存的核心数据需要大量关联查询的数据一句话总结核心业务数据放 MySQL、PostgreSQL、SQLite 等数据库中Redis 更适合放热点数据和临时数据。二十、完整练习代码最后给出一个可以直接运行的综合练习importjsonimportredisdefmain():rredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)print(连接状态,r.ping())print(\n String 示例 )r.set(app:name,Redis Python Demo,ex300)print(r.get(app:name))print(\n Counter 示例 )r.delete(article:1:view_count)for_inrange(5):r.incr(article:1:view_count)print(r.get(article:1:view_count))print(\n Hash 示例 )r.hset(user:1,mapping{name:Alice,age:20,city:Shanghai})print(r.hgetall(user:1))print(\n List 示例 )r.delete(user:1:recent)r.lpush(user:1:recent,article:100,article:101,article:102)print(r.lrange(user:1:recent,0,-1))print(\n Set 示例 )r.delete(article:1:likes)r.sadd(article:1:likes,user:1,user:2,user:1)print(r.smembers(article:1:likes))print(\n Sorted Set 示例 )r.delete(rank:score)r.zadd(rank:score,{Alice:98,Bob:86,Cindy:95})print(r.zrevrange(rank:score,0,-1,withscoresTrue))print(\n JSON 缓存示例 )user{id:1,name:Alice,skills:[Python,Redis]}r.set(cache:user:1,json.dumps(user,ensure_asciiFalse),ex60)cached_userjson.loads(r.get(cache:user:1))print(cached_user)if__name____main__:main()运行前确保Redis 服务已经启动已安装 Python 依赖pip install redis代码中的 host、port、password 配置正确总结Python 操作 Redis 的核心流程并不复杂安装 redis 库 - 连接 Redis - 选择数据结构 - 写入数据 - 读取数据 - 设置过期时间常用方法可以简单记住方法作用set()/get()读写字符串incr()/decr()计数器hset()/hgetall()操作 Hashlpush()/lrange()操作 Listsadd()/smembers()操作 Setzadd()/zrevrange()操作排行榜expire()/ttl()设置和查看过期时间delete()删除 key对 Python 初学者来说学习 Redis 的重点不是一开始就背所有命令而是先理解它适合解决什么问题。如果你能掌握缓存用户资料、验证码、计数器、排行榜这几个案例就已经可以在很多真实项目中使用 Redis 了。