Python实现Redis缓存MySQL数据并支持数据同步

简介

本文讲讲如何用Redis做MySQL的读缓存，提升数据库访问性能。

MySQL是一种很常用的关系型数据库，用于持久化数据，并存放在磁盘上。但如果有大数据量的读写，靠MySQL单点就会捉襟见肘，尽管可以在MySQL本身做优化，比如用更好的SQL语句设计、索引等等。也会用主从设计集群设计来优化性能。甚至借助工具做成分布式数据库。不过还有一种简单的方式来提升读性能，就是在MySQL的前面放一个缓存，比如Redis。Redis是一种高性能的内存数据库，用作缓存非常合适。Redis还支持分布式集群，来优化读写性能。Redis也可以持久化数据到磁盘，但Redis的持久化一定程度上会有丢数据的可能，因此数据完整性要求高的场合用MySQL更合适，而Redis用作缓存。

本文的重点是要解决数据的查询和更新过程中数据库的一致性问题。话不多说，开始上菜。

查询一致性

查询数据时：

先从Redis读取，如果存在则直接返回；
如果不存在则向MySQL查询数据；
1. 把从MySQL读取的数据更新到Redis；
2. 返回从MySQL读取的数据；

添加过期时间：

Redis的记录添加过期时间；
1. 如果没有记录，创建记录时会添加过期时间；
2. 如果有记录：
  1. 如果过期时间内没有被查询，自动被Redis删除数据；
  2. 如果过期时间内被查询，重置过期时间，续期；

这样可以定时清除Redis中查询不频繁的数据，增加数据读取速度。

更新一致性

数据更新时：

先查询redis，如果有数据，先删除缓存数据
然后先写入更新数据到redis，并设置过期时间
最后再写入更新数据到mysql：
1. 如果写入mysql失败，回滚（删除）redis和mysql的数据

MySQL表设计

使用学生表作为例子，存储学生的学号、姓名、生日、电话，主键为学号，建表语句如下：

# 以Linux命令行为例
mysql -uroot -pPASSWORD
CREATE DATABASE testdb;
USE testdb;
CREATE TABLE tb_student(stu_id INT AUTO_INCREMENT PRIMARY KEY NOT NULL,stu_name VARCHAR(20) NOT NULL,stu_birth DATE,stu_phone VARCHAR(100)
);

插入数据：

# 多插入几条不一样的数据
INSERT INTO tb_student(stu_name,stu_birth,stu_phone
) VALUES ('Tom','1900-11-11',15000000000
);

数据如下：

Redis数据设计

使用Redis的hash类型存储MySQL读缓存，因为hash类型一个表可以对应多个键值对：

hmset stu_id:1 stu_name 'Tom' stu_birth '1900-11-11' stu_phone '15000000000'

同时设置过期时间10分钟：

expire stu_id:1 600

Python代码

完整代码如下：

import pymysql
import redisclass DatabaseCache:def __init__(self):# 连接MySQLself.mysql = pymysql.connect(host='192.168.173.140',port=3306,user='root',password='123456',database='testdb',charset='utf8mb4')# 连接Redisself.redis = redis.Redis(host='192.168.173.140',port=6379,db=0,password='123456',decode_responses=True)def get_data(self, stu_id):# 根据学生id查询，name为redis hash表名name = f'stu_id:{stu_id}'res = self.redis.hgetall(name)# 查询到内容res为非空字典，否则为空字典if res:# 先查询Redis中是否存在数据，存在则直接返回，并且重置过期时间10分钟self.redis.expire(name, 600)print(f'get_data：redis有数据: {res}')return reselse:# 没有查询到数据with self.mysql.cursor() as cursor:try:# 从mysql查询cursor.execute(f'select stu_name, stu_birth, stu_phone from tb_student where stu_id={stu_id}')data = cursor.fetchall()print(f'get_data：redis无数据：mysql数据: {data}')# 把查询结果写入Redis中，并设置过期时间10分钟stu_name, stu_birth, stu_phone = data[0]  # 解包cache_data = {'stu_name': stu_name,'stu_birth': str(stu_birth),'stu_phone': stu_phone}self.redis.hset(name, mapping=cache_data)self.redis.expire(name, 600)  # 设置过期时间# 然后返回数据return dataexcept Exception as err:print(f'{err=}')def put_data(self, data: list):stu_id, stu_name, stu_birth, stu_phone = dataname = f'stu_id:{stu_id}'# 先查询redis中是否有数据，如果有先删除res = self.redis.hgetall(name)if res:# 如果只是部分更新，也可以不删除keys = self.redis.hkeys(name)self.redis.hdel(name, *keys)# 然后先写入新数据到redisnew_data = {'stu_name': stu_name,'stu_birth': stu_birth,'stu_phone': stu_phone}self.redis.hset(name, mapping=new_data)self.redis.expire(name, 600)# 更新完redis，再写入mysqlwith self.mysql.cursor() as cursor:try:# 执行失败会抛异常cursor.execute(f'select stu_id from tb_student where stu_id={stu_id}')_res = cursor.fetchone()  # 无记录返回Noneif _res:sql = f'update tb_student set stu_name="{stu_name}", stu_birth="{stu_birth}", ' \f'stu_phone="{stu_phone}" where stu_id={stu_id}'cursor.execute(sql)else:cursor.execute(f'insert into tb_student values ({stu_id}, "{stu_name}", "{stu_birth}", "{stu_phone}")')self.mysql.commit()print('redis有数据：mysql写入数据成功')except Exception as err:# 如果写入mysql不成功，需要回退redis，即删除redis中刚刚写入的数据keys = self.redis.hkeys(name)self.redis.hdel(name, *keys)self.mysql.rollback()  # mysql也要回退print(f'{err=}')def close(self):self.mysql.close()self.redis.close()if __name__ == '__main__':db = DatabaseCache()db.get_data(2)db.put_data([5, 'Eve', '1995-01-01', '18800003333'])db.close()