Python实现数据库定时任务自动重试机制
发布时间：2026-10-13| 2人浏览过
本文介绍了一个基于Python的数据库定时任务自动重试方案。针对下游依赖上游数据产出失败的场景，脚本通过循环检测与延时重试机制，自动执行SQL查询直至获取有效数据或达到最大重试次数，有效减少人工干预。
从事大数据处理的开发者经常需要编写定时任务来抽取数据。由于任务间存在依赖关系（通常是下游任务依赖上游任务的数据产出），这往往导致数据获取失败。当发现数据缺失时，
许多开发者会选择查看日志并手动触发任务执行。为此，本文实现了一个自动重试机制：脚本会持续尝试从数据库获取数据，若首次失败则自动重新执行，直到成功获取数据为止。
首先创建数据表：
CREATE TABLE `testtable` ( 2 `id` int(11) unsigned NOT NULL AUTO_INCREMENT, 3 `name` varchar(20) NOT NULL, 4 PRIMARY KEY (`id`) 5 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
初始状态下数据表为空，脚本在重试至第3秒时向数据库插入测试数据。
以下是Python代码的具体实现：
#!/usr/bin/env python #-*- coning:utf-8 -*- 3 4 import MySQLdb 5 from time import sleep 6 7 class GetData(object): 8 def __init__(self): 9 self.conn = '' 10 self.host = '127.0.0.1' 11 self.port = 3306 12 self.user = 'root' 13 self.passwd = '123456' 14 self.db = 'test' 15 self.cnum = 5 #set retry number 16 17 def init_connect(self): 18 self.conn = MySQLdb.connect(host=self.host, user=self.user, passwd=self.passwd, db=self.db, port=self.port, 19 charset='utf8') 20 21 def get_data(self): 22 self.init_connect 23 cur = self.conn.cursor 24 sql = "select * from testtable" 25 cur.execute(sql) 26 rs = cur.fetchall 27 cur.close 28 self.conn.close 29 return rs 30 31 def run(self): 32 count = 1 33 while (count <= self.cnum): 34 rs = self.get_data 35 if len(rs) > 0: 36 print len(rs) 37 break 38 39 print count 40 sleep(10) 41 count += 1 42 43 if __name__ == '__main__': 44 gd = GetData 45 gd.run
开发者可以手动执行操作，在代码运行至第3秒时，执行以下SQL语句：
insert into testtable(`name`) values ('123'),('456'),('789'),('1111'),('3222'),('444');
以下是配置定时任务的脚本示例：
00 08 * * * cd /home/python/lsh_sync; python getdata.py >> getdata.log 2>&1
以上介绍了使用Python编写的数据库定时任务自动重跑方案，希望能对各位有所帮助。如有疑问欢迎留言，小编将及时回复。感谢大家对本站的支持！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
