python-3.x 使用模块计划立即运行计划,然后每小时运行一次

xhv8bpkk  于 2023-07-01  发布在  Python
关注(0)|答案(5)|浏览(157)

我试图安排一个任务与模块“时间表”为每小时。我的问题是,我需要的任务,首先运行,然后再运行每小时。
这段代码工作正常,但在初始运行前要等待一个小时

import schedule
import time

def job():
    print("This happens every hour")

schedule.every().hour.do(job)

while True:
    schedule.run_pending()

我想避免这样做:

import schedule
import time

def job():
    print("This happens immediately then every hour")

schedule.every().hour.do(job)

while i == 0: 
    job()
    i = i+1

while i == 1:
    schedule.run_pending()

理想情况下,有这样一个选项会很好:

schedule.run_pending_now()
knpiaxh1

knpiaxh11#

可能最简单的解决方案是立即运行它并调度它,例如:

import schedule
import time

def job():
    print("This happens every hour")

schedule.every().hour.do(job)

job()                       # Runs now.
while True:
    schedule.run_pending()  # Runs every hour, starting one hour from now.
kdfy810k

kdfy810k2#

要运行所有作业,无论它们是否计划运行,请使用schedule.run_all()。作业在完成后被重新调度,就像使用run_pending()执行它们一样。

def job_1():
    print('Foo')

def job_2():
    print('Bar')

schedule.every().monday.at("12:40").do(job_1)
schedule.every().tuesday.at("16:40").do(job_2)

schedule.run_all()

# Add the delay_seconds argument to run the jobs with a number
# of seconds delay in between.
schedule.run_all(delay_seconds=10)```
7xllpg7q

7xllpg7q3#

如果有许多任务需要一些时间来执行,并且希望在启动期间独立运行它们,则可以使用threading

import schedule
import time

def job():
    print("This happens every hour")

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

run_threaded(job)       #runs job once during start

schedule.every().hour.do(run_threaded, job)
while True:
    schedule.run_pending()  # Runs every hour, starting one hour from now.
ar7v8xwq

ar7v8xwq4#

实际上,我不认为直接调用函数是明智的,因为它会阻塞没有调度器的线程,对吗?
我认为将作业设置为执行一次并没有错,例如每30秒执行一次:

scheduler.add_job(MPOStarter.run, args=ppi_args) # run once, then every 30 sec 
scheduler.add_job(MPOStarter.run, "interval", seconds=30, args=ppi_args)
rjee0c15

rjee0c155#

编写了一个实用程序来安排作业运行一次并按循环间隔运行。这比手动启动作业有优势,因为第一次运行将在主线程之外进行。

import schedule

from typing import Callable

def run_once_then_every(initial_delay: int, interval: int, job_func: Callable, *args, **kwargs):
    """
    Convenience wrapper to run a job once immediately (with some delay) then at a recurring interval.
    * Requires all intervals to be provided in minutes.
    * An initial delay of 0 results in the job running within 1 second.

    Note that the waiting period for the first periodic interval starts immediately and thus the
    interval between an initial call with a delay may be less than the interval between all other
    iterations.

    Args:
        - initial_delay (int): Delay before the first run of the job in minutes.
            An initial delay of 0 results in the job running within 1 second.
        - interval (int): Period between job recurrences.
        - job_func (Callable): The job to execute
        - args: args passed through to "job_func"
        - kwargs: kwargs passed through to "job_func"

    """

    def wrapped_job_func(*args, **kwargs):
        """
        Runs job_func once then returns a value indicating the scheduler
        should remove the job from its registry preventing future runs.
        """
        job_func(*args, **kwargs)
        return schedule.CancelJob

    # Schedule the first iteration
    if initial_delay == 0:
        # Close enough to immediate
        schedule.every(1).seconds.do(wrapped_job_func, *args, **kwargs)
    else:
        schedule.every(initial_delay).minutes.do(wrapped_job_func, *args, **kwargs)

    # Schedule periodic runs.
    schedule.every(interval).minutes.do(job_func, *args, **kwargs)

相关问题