1.安裝schedule,安裝命令如下:
pip install schedule
2.以下為普通簡(jiǎn)單的例子:
import schedule
import time
def job():
print("I'm working...")
# 每隔5分鐘執(zhí)行一次任務(wù)
schedule.every(5).minutes.do(job)
# 每隔一小時(shí)執(zhí)行一次任務(wù)
schedule.every().hour.do(job)
# 每天的10:30執(zhí)行一次任務(wù)
schedule.every().day.at("10:30").do(job)
# 每5-10分鐘執(zhí)行一次任務(wù)
schedule.every(5).to(10).minutes.do(job)
# 每周一的這個(gè)時(shí)候執(zhí)行一次任務(wù)
schedule.every().monday.do(job)
# 每周三的13:30執(zhí)行一次任務(wù)
schedule.every().wednesday.at("13:30").do(job)
# 每分鐘的這個(gè)時(shí)刻執(zhí)行一次任務(wù)
schedule.every().minute.at(":17").do(job)
# 如果job函數(shù)有有參數(shù)時(shí),這么寫(xiě)
schedule.every(10).seconds.do(job,"參數(shù)")
while True:
# 啟動(dòng)服務(wù)
run_pending()
#運(yùn)行所有可以運(yùn)行的任務(wù)
schedule.run_pending()
time.sleep(1)
3.schedule是串行執(zhí)行,如果時(shí)間重疊就會(huì)有重疊現(xiàn)象,但可以用多線程實(shí)現(xiàn)
import schedule
import time
import threading
def job():
print("I'm working... in job1 start")
time.sleep(15)
print("I'm working... in job1 end")
def job2():
print("I'm working... in job2")
def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()
#threading.Thread(target=loop,args=(i,loops[i]))
schedule.every(10).seconds.do(run_threaded,job)
schedule.every(10).seconds.do(run_threaded,job2)
while True:
schedule.run_pending()
time.sleep(1)
4.如果想要對(duì)線程的數(shù)量有所控制,則可以采用如下方法:
import Queue
import time
import threading
import schedule
def job():
print("I'm working")
def worker_main():
while 1:
job_func = jobqueue.get()
job_func()
jobqueue = Queue.Queue()
schedule.every(10).seconds.do(jobqueue.put, job)
schedule.every(10).seconds.do(jobqueue.put, job)
schedule.every(10).seconds.do(jobqueue.put, job)
schedule.every(10).seconds.do(jobqueue.put, job)
schedule.every(10).seconds.do(jobqueue.put, job)
worker_thread = threading.Thread(target=worker_main)
worker_thread.start()
while 1:
schedule.run_pending()
time.sleep(1)