mysql与redis结合使用

 综合利用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()复制