Python快速實現(xiàn)定時器的五種常見方法詳解

更新時間：2025年07月30日 10:17:03 作者：TIM老師

這篇文章主要為大家詳細介紹了使用Python語言快速實現(xiàn)定時器的五種常見方法,分為內(nèi)置方法與第三方庫的方法,感興趣的小伙伴可以了解下

1.sleep法（阻塞）

通過 while + sleep 實現(xiàn)定時任務

(1) 存在時間漂移（等待運行時間）

import time
def loopMonitor():
    while True:
        MonitorSystem()
        # 1min檢查一次
        time.sleep(60)
loopMonitor()

(2) 維護sleep時間

import time
def loopMonitor():
    while True:
        MonitorSystem()
        #2s檢查一次
        time.sleep(60.0 - ((time.time() - starttime) % 60.0))
loopMonitor()

2.自定義法（+線程、非阻塞）

import threading 
import time
 
class RepeatedTimer(object):
  def __init__(self, interval, function, *args, **kwargs):
    self._timer = None
    self.interval = interval
    self.function = function
    self.args = args
    self.kwargs = kwargs
    self.is_running = False
    self.next_call = time.time()
    self.start()
 
  def _run(self):
    self.is_running = False
    self.start()
    self.function(*self.args, **self.kwargs)
 
  def start(self):
    if not self.is_running:
      self.next_call += self.interval
      self._timer = threading.Timer(self.next_call - time.time(), self._run)
      self._timer.start()
      self.is_running = True
 
  def stop(self):
    self._timer.cancel()
    self.is_running = False
    
    
from time import sleep
def hello(name):
    print "Hello %s!" % name
 
print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
    sleep(5) # your long-running job goes here...
finally:
    rt.stop() # better in a try/finally block to make sure the program ends!

3.Twisted

更加健壯，并且實現(xiàn)了許多功能：守護進程、日志記錄或異常處理

from twisted.internet import task, reactor
 
timeout = 60.0 # Sixty seconds
 
def doWork():
    #do work here
    pass
 
l = task.LoopingCall(doWork)
l.start(timeout) # call every sixty seconds
 
reactor.run()

4.schedule （簡單方便，推薦）

import schedule
import time
 
def job():
    print("I'm working...")
 
#schedule.every(1)創(chuàng)建Job, seconds.do(func)按秒間隔查詢并執(zhí)行
schedule.every(1).seconds.do(func)
#添加任務按分執(zhí)行
schedule.every(1).minutes.do(func)
#添加任務按天執(zhí)行
schedule.every(1).days.do(func)
#添加任務按周執(zhí)行
schedule.every().weeks.do(func)
#添加任務每周一執(zhí)行，執(zhí)行時間為下周一這一時刻時間
schedule.every().monday.do(func)
 
 
schedule.every().monday.at("12:00").do(job)
# 每隔10分鐘執(zhí)行一次任務
schedule.every(10).minutes.do(job)
# 每隔一小時執(zhí)行一次任務
schedule.every().hour.do(job)
# 每天10:30執(zhí)行一次任務
schedule.every().day.at("10:30").do(job)
# 每隔5到10分鐘運行一次任務？
schedule.every(5).to(10).minutes.do(job)
# 每周一的這個時候執(zhí)行一次任務
schedule.every().monday.do(job)
# 每周三 13:15執(zhí)行一次任務
schedule.every().wednesday.at("13:15").do(job)
# # 每分鐘的第17秒執(zhí)行任務  
schedule.every().minute.at(":17").do(job)
 
while True:
    schedule.run_pending()   # 檢測是否有到期任務
    time.sleep(1)  # 避免一直查詢cpu過高。sleep時間不應該小于執(zhí)行時間，否則會線程堆積。

5.Apscheduler （功能多，推薦）

調(diào)度器（scheduler）

BlockingScheduler: 調(diào)度器在當前進程的主線程中運行，會阻塞當前線程。

BackgroundScheduler: 調(diào)度器在后臺線程中運行，不會阻塞當前線程。

AsyncIOScheduler: 結(jié)合asyncio模塊一起使用。

GeventScheduler: 程序中使用gevent作為IO模型和GeventExecutor配合使用。

TornadoScheduler: 程序中使用Tornado的IO模型，用 ioloop.add_timeout 完成定時喚醒。

TwistedScheduler: 配合TwistedExecutor，用reactor.callLater完成定時喚醒。

QtScheduler: 應用是一個Qt應用，需使用QTimer完成定時喚醒。

觸發(fā)器（trigger）

date是最基本的一種調(diào)度，作業(yè)任務只會執(zhí)行一次。

interval觸發(fā)器，固定時間間隔觸發(fā)。

cron 觸發(fā)器，在特定時間周期性地觸發(fā)，和Linux crontab格式兼容。它是功能最強大的觸發(fā)器。

作業(yè)存儲（job store）

添加任務，有兩種添加方法，一種add_job()，另一種是scheduled_job()修飾器來修飾函數(shù)。

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler 
scheduler = BlockingScheduler() 
 
# 第一種 
@scheduler.scheduled_job(job_func, 'interval', seconds=10) 
def timed_task():     
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) 
    
# 第二種 
scheduler.add_job(timed_task, 'interval', seconds=5) 
scheduler.start()

刪除任務，兩種方法：remove_job() 和 job.remove()。remove_job()是根據(jù)任務的id來移除，所以要在任務創(chuàng)建的時候指定一個 id。job.remove()則是對任務執(zhí)行remove方法。

scheduler.add_job(job_func, 'interval', seconds=20, id='one') 
scheduler.remove_job(one) 
 
task = add_job(task_func, 'interval', seconds=2, id='job_one') 
task.remvoe()

獲取任務列表，通過scheduler.get_jobs()方法能夠獲取當前調(diào)度器中的所有任務的列表

tasks = scheduler.get_jobs()

關閉任務，使用scheduler.shutdown()默認情況下調(diào)度器會等待所有正在運行的作業(yè)完成后，關閉所有的調(diào)度器和作業(yè)存儲。

scheduler.shutdown() 
scheduler.shutdown(wait=false)

注意點

時區(qū)問題

報錯：ValueError: Timezone offset does not match system offset: 0 != 28800. Please, check your config files.

分析：通過分析異常日志，發(fā)現(xiàn)APScheduler的默認timezone，而“0”是獲取的系統(tǒng)環(huán)境變量的TZ時間28800對應timezone為“Asia/Shanghai”，而0對應timezone為“UTC”，所以需將系統(tǒng)環(huán)境變量的時區(qū)與APScheduler的時區(qū)設置為一致。

解決：

#!/usr/bin/python
# -*- coding: utf-8 -*-
 
from apscheduler.schedulers.background import BackgroundScheduler    
import os
 
os.environ['TZ']= "Asia/Shanghai"
scheduler = BackgroundScheduler(timezone="Asia/Shanghai")

其他解決方案：

如果部署應用dockerfile配置，也可以在dockerfile中設定系統(tǒng)時區(qū)。

到此這篇關于Python快速實現(xiàn)定時器的五種常見方法詳解的文章就介紹到這了,更多相關Python定時器內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: