当前位置： 移动技术网 > IT编程>脚本编程>Python > python:定时任务模块schedule

python:定时任务模块schedule

2019年02月26日  | 移动技术网IT编程  | 我要评论

玲声,格瑞特妖怪学院txt,江潮电脑维修

1.安装

pip install schedule

2.文档

https://schedule.readthedocs.io/en/stable/faq.html#how-to-execute-jobs-in-parallel

3.官网使用demo

import schedule
import time

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)

4.我的schedule使用demo

import sys
import time
import schedule
import os
import logging
if not os.path.exists('/var/log/video_download/'):
    os.makedirs('/var/log/video_download')
log = logging.getlogger()
log.setlevel(logging.debug)
fmt = logging.formatter("%(asctime)s %(pathname)s %(filename)s %(funcname)s %(lineno)s %(levelname)s - %(message)s",
                        "%y-%m-%d %h:%m:%s")
stream_handler = logging.filehandler(
    '/var/log/video_download/debug-%s.log' % (time.strftime('%y-%m-%d', time.localtime(time.time()))))
stream_handler.setlevel(logging.debug)
stream_handler.setformatter(fmt)
log.addhandler(stream_handler)
def handler():
    print("this is handler")
def main():
    if len(sys.argv) != 2:
        print('python video_data.py hour')
        sys.exit()
    param = sys.argv[1]
    if param == 'hour':
        log.debug("enter main")
        schedule.every().day.at("00:00").do(handler)
        schedule.every().hour.do(handler)
        while true:
            schedule.run_pending()
            time.sleep(1)
    else:
        print("python video_data.py hour")
        sys.exit()


if __name__ == "__main__":
    main()

 

5.拓展：

　　并行执行任务

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

import threading
import time
import 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)
schedule.every(10).seconds.do(run_threaded, job)
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

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.task_done()

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)

　　(3)抛出异常

import functools

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)只运行一次

def job_that_executes_once():
    # do some work ...
    return schedule.canceljob

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

　　(5)一次取消多个任务

def greet(name):
    print('hello {}'.format(name))

schedule.every().day.do(greet, 'andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'john').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'derek').tag('daily-tasks', 'guest')

schedule.clear('daily-tasks')

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

import functools
import time

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)随机开展工作

def my_job():
    # this job will execute every 5 to 10 seconds.
    print('foo')

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

　　(8)传参给作业函数

def greet(name):
    print('hello', name)

schedule.every(2).seconds.do(greet, name='alice')
schedule.every(4).seconds.do(greet, name='bob')

您可能感兴趣的文章:

如对本文有疑问，请在下面进行留言讨论，广大热心网友会与你互动！！ 点击进行留言回复

相关文章:

验证码：
最近更新的文章
大家感兴趣的文章
移动技术网