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

news2024/11/26 6:22:04

简介

本文讲讲如何用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 redis


class DatabaseCache:
    def __init__(self):
        # 连接MySQL
        self.mysql = pymysql.connect(
            host='192.168.173.140',
            port=3306,
            user='root',
            password='123456',
            database='testdb',
            charset='utf8mb4'
        )

        # 连接Redis
        self.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 res

        else:
            # 没有查询到数据
            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 data

                except Exception as err:
                    print(f'{err=}')

    def put_data(self, data: list):
        stu_id, stu_name, stu_birth, stu_phone = data
        name = f'stu_id:{stu_id}'

        # 先查询redis中是否有数据，如果有先删除
        res = self.redis.hgetall(name)
        if res:
            # 如果只是部分更新，也可以不删除
            keys = self.redis.hkeys(name)
            self.redis.hdel(name, *keys)

        # 然后先写入新数据到redis
        new_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，再写入mysql
        with self.mysql.cursor() as cursor:
            try:
                # 执行失败会抛异常
                cursor.execute(f'select stu_id from tb_student where stu_id={stu_id}')
                _res = cursor.fetchone()  # 无记录返回None
                if _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()