python:定时任务模块schedule

发布时间：2020-12-17 00:01:38 所属栏目：Python 来源：网络整理

导读：1.安装 pip install schedule 2.官网使用demo span style="color: #0000ff;"def span style="color: #000000;" job(): span style="color: #0000ff;"print (span style="color: #800000;"" span style="color: #800000;"I'm working... span style="color: #

1.安装

pip install schedule

2.官网使用demo

def job():
print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

while True:
schedule.run_pending()
time.sleep(1)

3.拓展：

　　1.并行执行任务

　　(1)默认情况下，schedule按顺序执行所有作业。这背后的原因是很难找到一个让每个人都开心的并行执行模型

def job():
print("I'm running on thread %s" % threading.current_thread())

def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()

schedule.every(10).seconds.do(run_threaded,job)
schedule.every(10).seconds.do(run_threaded,job)

while 1:
schedule.run_pending()
time.sleep(1)

　　(2)如果需要控制线程数，就需要用queue

def job():
print("I'm working")

def worker_main():
while 1:
job_func = jobqueue.get()
job_func()
jobqueue.task_done()

jobqueue = Queue.Queue()

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)

　　(3)抛出异常

def catch_exceptions(cancel_on_failure=False):
def catch_exceptions_decorator(job_func):
@functools.wraps(job_func)
def wrapper(*args,*kwargs):
try:
return job_func(args,**kwargs)
except:
import traceback
print(traceback.format_exc())
if cancel_on_failure:
return schedule.CancelJob
return wrapper
return catch_exceptions_decorator

@catch_exceptions(cancel_on_failure=True)
def bad_task():
return 1 / 0

schedule.every(5).minutes.do(bad_task)

　　(4)只运行一次

schedule.every().day.at('22:30').do(job_that_executes_once)

　　(5)一次取消多个任务

(schedule.every().day.do(greet,'Andrea').tag('daily-tasks','friend')
schedule.every().hour.do(greet,'John').tag('hourly-tasks','Monica').tag('hourly-tasks','customer')
schedule.every().day.do(greet,'Derek').tag('daily-tasks','guest')

schedule.clear('daily-tasks')

　　(6)在任务中加入日志功能

import schedule

# This decorator can be applied to
def with_logging(func):
@functools.wraps(func)
def wrapper(*args,*kwargs):
print('LOG: Running job "%s"' % func.name)
result = func(args,**kwargs)
print('LOG: Job "%s" completed' % func.name)
return result
return wrapper

@with_logging
def job():
print('Hello,World.')

schedule.every(3).seconds.do(job)

while 1:
schedule.run_pending()
time.sleep(1)

　　(7)随机开展工作

(schedule.every(5).to(10).seconds.do(my_job)

　　(8)传参给作业函数

(schedule.every(2).seconds.do(greet,name='Alice')
schedule.every(4).seconds.do(greet,name='Bob')

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!