Python定時從Mysql提取數據存入Redis的實現

設計思路：
1.程序一旦run起來，python會把mysql中最近一段時間的數據全部提取出來
2.然後實例化redis類，將數據簡單解析後逐條傳入redis隊列
3.定時器設計每天凌晨12點開始跑
ps：redis是個內存數據庫，做後臺消息隊列的緩存時有很大的用處，有興趣的小夥伴可以去查看相關的文檔。
# -*- coding:utf-8 -*- import MySQLdb import schedule import time import datetime import random import string import redis # get the data from mysql class FromSql(object): def __init__(self, conn): self.conn = conn def acquire(self): cursor = self.conn.cursor() try: sql = "SELECT * FROM test WHERE TO_DAYS(NOW()) - TO_DAYS(t) <= 1" cursor.execute(sql) rs = cursor.fetchall() #print (rs) for eve in rs: print('%s, %s, %s, %s' % eve) copy_rs = rs cursor.close() return copy_rs except Exception as e: print("The error: %s" % e) class RedisQueue(object): def __init__(self, name, namespace='queue', **redis_kwargs): """The default connection parameters are: host='localhost', port=6379, db=0""" self.__db= redis.Redis(**redis_kwargs) self.key = '%s:%s' %(namespace, name) def qsize(self): return self.__db.llen(self.key) def put(self, item): self.__db.rpush(self.key, item) def get(self, block=True, timeout=None): if block: item = self.__db.blpop(self.key, timeout=timeout) else: item = self.__db.lpop(self.key) if item: item = item[1] return item def get_nowait(self): return self.get(False) if __name__ == "__main__": # connect mysqldb conn_sql = MySQLdb.connect( host = '127.0.0.1', port = 3306, user = 'root', passwd = '', db = 'test', charset = 'utf8' ) def job_for_redis(): get_data = FromSql(conn_sql) data = get_data.acquire() q = RedisQueue('test',host='localhost', port=6379, db=0) for single_data in data: for meta_data in single_data: q.put(meta_data) print(meta_data) print("All data had been inserted.") """ try: schedule.every().day.at("00:00").do(job_for_redis) except Exception as e: print('Error: %s'% e) # finally: # conn.close() while True: schedule.run_pending() time.sleep(1) """
補充知識：python定時獲取匯率存入數據庫
python定時任務：
我們可以使用 輕量級的第三方模塊schedule。首先先安裝：pip install schedule
定時任務的的小測試：
import schedule import time def job(): print("I'm working...") schedule.every(10).minutes.do(job) # 每隔10分鐘執行一次任務 schedule.every().hour.do(job) # 每隔一小時執行一次任務 schedule.every().day.at("10:30").do(job) # 每天10：30執行一次任務 schedule.every(5).to(10).days.do(job) # 每5-10天執行一次任務 schedule.every().monday.do(job) # 每週一的這個時候執行一次任務 schedule.every().wednesday.at("13:15").do(job) # 每週三13:15執行一次任務 while True: schedule.run_pending()
獲取數據存入數據庫：（格式可能不太對，還有一些符號。自己修改一下即可）
import pymysql import schedule import time import requests import pandas from sqlalchemy import create_engine #獲取美元的所有外匯 def job(): content = '美元' url = 'http://www.boc.cn/sourcedb/whpj/index.html' #外匯數據地址 html = requests.get(url).content.decode('utf-8') index = html.index('') str = html[index:index+300] result = re.findall('',str) print("幣種：" + result[0]) print("現匯買入價：" + result[1]) print("現鈔買入價：" + result[2]) print("現匯賣出價：" + result[3]) print("現鈔賣出價：" + result[4]) print("中行結算價：" + result[5]) print("發佈時間：" + result[6] + ' ' + result[7]) #本地地址 數據庫賬號 密碼 數據庫名 db = pymysql.connect('localhost','root','root','pinyougoudb') cursor = db.cursor() #sql語句 sql = "update tb_money set huiBuy = %s,chaoBuy = %s,huiSale = %s,chaoSale = %s,centerResult= %s,publishTime = '%s' where typeId = '%s'" % (result[1], result[2], result[3], result[4], result[5], result[6] + ' ' + result[7], result[0]) cursor.execute(sql) db.commit() print('success') # 查詢語句，將存入的數據查出來 # sqlalchemy 進行數據庫初始化 engine = create_engine('mysql+pymysql://root:root@localhost:3306/pinyougoudb') sql = '''select * from tb_money''' # pandas 進行數據庫讀寫 df = pandas.read_sql_query(sql,engine) print(df) db.commit() # 每隔幾分中刷新一次 #schedule.every(0.1).minutes.do(job) #每天什麼時候刷新 schedule.every().day.at("09:29").do(job) schedule.every().day.at("09:30").do(job) #一直循環 知道滿足條件執行 while True: schedule.run_pending()