使用threading写的一个定时器任务demo:
- import time
- import sys
- import signal
- import datetime
- import threading
- #定时器
- def schedule_update():
- t = threading.Timer(0, event_func)
- t.setDaemon(True)
- t.start()
- #执行函数
- def event_func():
- now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
- print(now_time)
- exec_update()
- #update_openvas_dbs_from_cache()
- interval_time = delay_time()
- t = threading.Timer(interval_time, event_func)
- t.setDaemon(True)
- t.start()
- #取时间点
- def delay_time():
- # now time
- now_time = datetime.datetime.now()
- # tomorrow time
- next_time = now_time + datetime.timedelta(days=+1)
- next_year = next_time.date().year
- next_month = next_time.date().month
- next_day = next_time.date().day
- # get tomorrow 00:00
- next_time = datetime.datetime.strptime(str(next_year)+"-"+str(next_month)+"-"+str(next_day)+" 00:00:00", "%Y-%m-%d %H:%M:%S")
- # get secondes
- delay_time = (next_time - now_time).total_seconds()
- return delay_time
- def quit_sys(signum, frame):
- sys.exit()
- #接收C
- if __name__ == "__main__":
- try:
- signal.signal(signal.SIGINT, quit_sys)
- signal.signal(signal.SIGTERM, quit_sys)
- schedule_update()
- print("schedule_update server starting up...\nHit Ctrl-C to quit.\n")
- while 1:
- time.sleep(1)
- except Exception as e:
- print(e)
总结
以上所述是小编给大家介绍的Python3定时器任务代码,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对w3xue网站的支持!
如果你觉得本文对你有帮助,欢迎转载,烦请注明出处,谢谢!