综合利用redis内存数据库和mysql关系型数据库特点，提升系统性能
核心思想：先从redis读，redis没有在从mysql读，读到后放入redis，写的时间同时写入redis与mysql，前台写入redis， 在后台写入mysql
import pymysql
import redis
class MySqlHelper:
def __init__(self):
self.con = None
self.cur = None
try:
self.con = pymysql.connect(user="root", password="123456", database="python2407")
except Exception as e:
print(f"连接出异常")
else:
self.cur = self.con.cursor()
def query_one(self, sql, args=None):
try:
self.cur.execute(sql, args)
return self.cur.fetchone()
except Exception as e:
print(f"查询出异常")
def update(self, sql, args=None):
try:
row = self.cur.execute(sql, args)
self.con.commit()
return row
except Exception as e:
print(f"修改出异常")
def __del__(self):
if self.cur:
self.cur.close()
if self.con:
self.con.close()
class RedisHelper:
def __init__(self):
self.client = None
try:
self.client = redis.Redis(host="192.168.13.32", password="123456")
except Exception as e:
print(f"连接redis失败")
def query_one(self, key):
try:
return self.client.hgetall(key)
except Exception as e:
print(f"执行查询失败")
def update(self, key, value):
try:
self.client.hset(key, mapping=value)
except Exception as e:
print(f"执行修改失败", e)
def __del__(self):
if self.client:
self.client.close()
class DataHelper:
def __init__(self):
self.sql = MySqlHelper()
self.redis = RedisHelper()
def regist_user(self, username, password):
row = self.sql.update("insert into user (username, password) values (%s, %s)", (username, password))
if row > 0:
user = self.sql.query_one("select * from user where username = %s and password = %s", (username, password))
self.redis.update(f"{user[0]}", {"id": user[0], "username": user[1], "password": user[2]})
print(f"注册成功")
def query_user(self, u_id):
value = self.redis.query_one(u_id)
if value:
print(f"redis中直接找到")
return value
else:
user = self.sql.query_one("select * from user where id = %s", u_id)
if user:
print(f"redis中没有找到， 从mysql中找到放入redis")
self.redis.update(f"{u_id}", {"id": user[0], "username": user[1], "password": user[2]})
return user
def main():
dh = DataHelper()
r = dh.query_user("3031")
print(r)
# dh.regist_user("zzy1", "123")
if __name__ == '__main__':
main()