快捷導(dǎo)航

python BlockingScheduler定時(shí)任務(wù)及其他方式的實(shí)現(xiàn)

更新時(shí)間：2019年09月19日 14:54:35 作者：白熊花田

這篇文章主要介紹了python BlockingScheduler定時(shí)任務(wù)及其他方式的實(shí)現(xiàn)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

本文介紹了python BlockingScheduler定時(shí)任務(wù)及其他方式的實(shí)現(xiàn)，具體如下：

#BlockingScheduler定時(shí)任務(wù)
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
首先看看周一到周五定時(shí)執(zhí)行任務(wù)
# 輸出時(shí)間
def job():
  print(datetime.now().strtime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, "cron"， day_of_week="1-5", hour=6, minute=30)
schduler.start()


scheduler.add_job(job, 'cron', hour=1, minute=5)
hour =19 , minute =23  這里表示每天的19：23 分執(zhí)行任務(wù)
hour ='19', minute ='23'  這里可以填寫(xiě)數(shù)字，也可以填寫(xiě)字符串
hour ='19-21', minute= '23'  表示 19:23、 20:23、 21:23 各執(zhí)行一次任務(wù)

#每300秒執(zhí)行一次
scheduler .add_job(job, 'interval', seconds=300)

#在1月,3月,5月,7-9月，每天的下午2點(diǎn)，每一分鐘執(zhí)行一次任務(wù)
scheduler .add_job(func=job, trigger='cron', month='1,3,5,7-9', day='*', hour='14', minute='*')

# 當(dāng)前任務(wù)會(huì)在 6、7、8、11、12 月的第三個(gè)周五的 0、1、2、3 點(diǎn)執(zhí)行
scheduler .add_job(job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')

#從開(kāi)始時(shí)間到結(jié)束時(shí)間，每隔倆小時(shí)運(yùn)行一次
scheduler .add_job(job, 'interval', hours=2, start_date='2018-01-10 09:30:00', end_date='2018-06-15 11:00:00')

#自制定時(shí)器
 from datetime import datetime
 import time
 # 每n秒執(zhí)行一次
 def timer(n):
   while True:
     print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
     time.sleep(n)

timer(5)

Python下實(shí)現(xiàn)定時(shí)任務(wù)的方式有很多種方式。

循環(huán)sleep：

這是一種最簡(jiǎn)單的方式，在循環(huán)里放入要執(zhí)行的任務(wù)，然后sleep一段時(shí)間再執(zhí)行。缺點(diǎn)是，不容易控制，而且sleep是個(gè)阻塞函數(shù)。

def timer(n):
  '''
  每n秒執(zhí)行一次
  '''
  while True: 
    print time.strftime('%Y-%m-%d %X',time.localtime()) 
    yourTask() # 此處為要執(zhí)行的任務(wù) 
    time.sleep(n)

threading的Timer：

threading模塊中的Timer能夠幫助實(shí)現(xiàn)定時(shí)任務(wù)，而且是非阻塞的。

比如3秒后打印helloworld：

def printHello():
  print "hello world"
 
 
Timer(3, printHello).start()

比如每3秒打印一次helloworld：

def printHello():
  print "Hello World"
  t = Timer(2, printHello)
  t.start()
 
 
if __name__ == "__main__":
  printHello()

使用sched模塊：

sched是一種調(diào)度（延時(shí)處理機(jī)制）。

# -*- coding:utf-8 -*-
# use sched to timing
import time
import os
import sched
 
 
# 初始化sched模塊的scheduler類(lèi)
# 第一個(gè)參數(shù)是一個(gè)可以返回時(shí)間戳的函數(shù)，第二個(gè)參數(shù)可以在定時(shí)未到達(dá)之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
 
 
# 被周期性調(diào)度觸發(fā)的函數(shù)
def execute_command(cmd, inc):
  '''
  終端上顯示當(dāng)前計(jì)算機(jī)的連接情況
  '''
  os.system(cmd)
  schedule.enter(inc, 0, execute_command, (cmd, inc))
 
 
def main(cmd, inc=60):
  # enter四個(gè)參數(shù)分別為：間隔事件、優(yōu)先級(jí)（用于同時(shí)間到達(dá)的兩個(gè)事件同時(shí)執(zhí)行時(shí)定序）、被調(diào)用觸發(fā)的函數(shù)，
  # 給該觸發(fā)函數(shù)的參數(shù)（tuple形式）
  schedule.enter(0, 0, execute_command, (cmd, inc))
  schedule.run()
 
 
# 每60秒查看下網(wǎng)絡(luò)連接情況
if __name__ == '__main__':
  main("netstat -an", 60)

使用定時(shí)框架APScheduler：

APScheduler是基于Quartz的一個(gè)Python定時(shí)任務(wù)框架。提供了基于日期、固定時(shí)間間隔以及crontab類(lèi)型的任務(wù)，并且可以持久化任務(wù)。

這個(gè)現(xiàn)在還沒(méi)自己嘗試過(guò)，等過(guò)段時(shí)間用了再來(lái)補(bǔ)充。

使用windows的定時(shí)任務(wù)：

這里可以將所需要的Python程序打包成exe文件，然后在windows下設(shè)置定時(shí)執(zhí)行。

使用Linux的定時(shí)任務(wù)（Crontab）：

在Linux下可以很方便的借助Crontab來(lái)設(shè)置和運(yùn)行定時(shí)任務(wù)。進(jìn)入Crontab文件編輯頁(yè)面，設(shè)置時(shí)間間隔，使用一些shell命令來(lái)運(yùn)行bash腳本或者是Python腳本，保存后Linux會(huì)自動(dòng)按照設(shè)定的時(shí)間來(lái)定時(shí)運(yùn)行程序。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: