在 Python 3 中安排重复事件

Question

我正在尝试安排一个重复事件在 Python 3 中每分钟运行一次。

我已经看到了 sched.scheduler 类，但我想知道是否还有其他方法可以做到这一点。我听说我可以为此使用多个线程，我不介意这样做。

我基本上是请求一些 JSON，然后解析它；它的价值随着时间的推移而变化。

要使用 sched.scheduler，我必须创建一个循环来请求它安排事件运行一小时：

scheduler = sched.scheduler(time.time, time.sleep)

# Schedule the event. THIS IS UGLY!
for i in range(60):
    scheduler.enter(3600 * i, 1, query_rate_limit, ())

scheduler.run()

还有哪些其他方法可以做到这一点？

Answer 1

采用原始的 threading.Timer() 类实现并修复 run() 方法，我得到类似以下内容：

class PeriodicTimer(Thread):
    """A periodic timer that runs indefinitely until cancel() is called."""
    def __init__(self, interval, function, args=None, kwargs=None):
        Thread.__init__(self)
        self.interval = interval
        self.function = function
        self.args = args if args is not None else []
        self.kwargs = kwargs if kwargs is not None else {}
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet."""
        self.finished.set()

    def run(self):
        """Run until canceled"""
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)

调用的 wait() 方法是使用条件变量，所以它应该相当有效。

Answer 2

不久前我遇到了类似的问题，所以我制作了一个 python 模块 event-scheduler 来解决这。它具有与 sched 库非常相似的 API，但有一些差异：

1. 它使用后台线程，并且始终能够在后台接受和运行作业，直到调度程序显式停止（不需要 while 循环）。
2. 它附带一个 API，可以按照用户指定的时间间隔安排重复事件，直到明确取消为止。

可以通过pip install event-scheduler安装

from event_scheduler import EventScheduler

event_scheduler = EventScheduler()
event_scheduler.start()
# Schedule the recurring event to print "hello world" every 60 seconds with priority 1
# You can use the event_id to cancel the recurring event later
event_id = event_scheduler.enter_recurring(60, 1, print, ("hello world",))

Answer 3

我从 JavaScript 中找到了 setTimeout 和 setInterval 的模仿，但是是在 Python 中。

使用此类，您可以：

1. 使用间隔 ID 设置超时（通过变量，不是必需的）。
2. 以与超时相同的方式设置间隔。
3. 使用 ID 设置间隔，然后设置超时，该超时将在给定时间内取消该间隔。
4. 用一个ID设置超时，用另一个setTimeout取消超时。

我知道它看起来很复杂，但我需要它拥有的所有功能，就像 JavaScript 中一样。请注意，您需要预先创建一个函数并使用函数名称作为参数（请参阅代码）。

#If I'm not mistaken, I believe the only modules you need are time and threading. The rest were for something I'm working on.
import time
import math
import pygame
import threading
import ctypes
from sys import exit

#Functions
class Timer:
    def __init__(self):
        self.timers = {}
        self.timer_id = 0
    
    def setInterval(self, fn, time, *args):
        def interval_callback():
            fn(*args)
            if timer_id in self.timers:
                self.timers[timer_id] = threading.Timer(time/1000, interval_callback)
                self.timers[timer_id].start()

        timer_id = self.timer_id
        self.timer_id += 1
        self.timers[timer_id] = threading.Timer(time/1000, interval_callback)
        self.timers[timer_id].start()
        return timer_id

    def clearInterval(self, timer_id):
        if timer_id in self.timers:
            self.timers[timer_id].cancel()
            del self.timers[timer_id]

    def setTimeout(self, fn, delay, *args, **kwargs):
        def timer_callback():
            self.timers.pop(timer_id, None)
            fn(*args, **kwargs)
        
        timer_id = self.timer_id
        self.timer_id += 1
        t = threading.Timer(delay / 1000, timer_callback)
        self.timers[timer_id] = t
        t.start()
        return timer_id
    
    def clearTimeout(self, timer_id):
        t = self.timers.pop(timer_id, None)
        if t is not None:
            t.cancel()
t = Timer()

#Usage Example
lall = t.setInterval(print, 1000, "hi")
t.setTimeout(t.clearInterval, 3000, lall)

希望这有帮助。

Answer 4

您可以使用 threading.Timer ，但这也会调度一次性事件，类似于调度程序对象的 .enter 方法。

将一次性调度程序转换为周期性调度程序的正常模式（在任何语言中）是让每个事件以指定的时间间隔重新调度自身。例如，对于 sched，我不会像您一样使用循环，而是使用类似的东西：

def periodic(scheduler, interval, action, actionargs=()):
    scheduler.enter(interval, 1, periodic,
                    (scheduler, interval, action, actionargs))
    action(*actionargs)

并通过调用启动整个“永久定期计划”

periodic(scheduler, 3600, query_rate_limit)

或者，我可以使用 threading.Timer 而不是 scheduler.enter，但模式非常相似。

如果您需要更精细的变化（例如，在给定时间或在某些条件下停止定期重新安排），那么使用一些额外参数并不难。

Answer 5

您可以使用时间表。它适用于 Python 2.7 和 3.3，并且相当轻量级：

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)

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

Answer 6

我对这个主题的粗浅看法：

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.function   = function
        self.interval   = interval
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

用法：

from time import sleep

def hello(name):
    print "Hello %s!" % name

print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
    sleep(5) # your long-running job goes here...
finally:
    rt.stop() # better in a try/finally block to make sure the program ends!

功能：

• 仅标准库，无外部依赖项
• 使用 Alex Martnelli 建议的模式
• start() 和 stop() 可以安全调用多次即使计时器已经启动/停止
• 要调用的函数可以有位置和命名参数
• 您可以随时更改间隔，它将在下次运行后生效。对于 args、kwargs 甚至 function 也是如此！

Answer 7

基于MestreLion的回答，它解决了多线程的一个小问题：

from threading import Timer, Lock

class Periodic(object):
    """
    A periodic task running in threading.Timers
    """

    def __init__(self, interval, function, *args, **kwargs):
        self._lock = Lock()
        self._timer = None
        self.function = function
        self.interval = interval
        self.args = args
        self.kwargs = kwargs
        self._stopped = True
        if kwargs.pop('autostart', True):
            self.start()

    def start(self, from_run=False):
        self._lock.acquire()
        if from_run or self._stopped:
            self._stopped = False
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
        self._lock.release()

    def _run(self):
        self.start(from_run=True)
        self.function(*self.args, **self.kwargs)

    def stop(self):
        self._lock.acquire()
        self._stopped = True
        self._timer.cancel()
        self._lock.release()

Answer 8

您可以使用高级 Python 调度程序。它甚至有一个类似 cron 的界面。

Answer 9

使用 Celery。

from celery.task import PeriodicTask
from datetime import timedelta


class ProcessClicksTask(PeriodicTask):
    run_every = timedelta(minutes=30)

    def run(self, **kwargs):
        #do something

Answer 10

根据 Alex Martelli 的回答，我实现了更容易集成的装饰器版本。

import sched
import time
import datetime
from functools import wraps
from threading import Thread


def async(func):
    @wraps(func)
    def async_func(*args, **kwargs):
        func_hl = Thread(target=func, args=args, kwargs=kwargs)
        func_hl.start()
        return func_hl
    return async_func


def schedule(interval):
    def decorator(func):
        def periodic(scheduler, interval, action, actionargs=()):
            scheduler.enter(interval, 1, periodic,
                            (scheduler, interval, action, actionargs))
            action(*actionargs)

        @wraps(func)
        def wrap(*args, **kwargs):
            scheduler = sched.scheduler(time.time, time.sleep)
            periodic(scheduler, interval, func)
            scheduler.run()
        return wrap
    return decorator


@async
@schedule(1)
def periodic_event():
    print(datetime.datetime.now())


if __name__ == '__main__':
    print('start')
    periodic_event()
    print('end')

Answer 11

文档：高级 Python 调度程序

@sched.cron_schedule(day='last sun')
def some_decorated_task():
    print("I am printed at 00:00:00 on the last Sunday of every month!")

可用字段：

| Field       | Description                                                    |
|-------------|----------------------------------------------------------------|
| year        | 4-digit year number                                            |
| month       | month number (1-12)                                            |
| day         | day of the month (1-31)                                        |
| week        | ISO week number (1-53)                                         |
| day_of_week | number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) |
| hour        | hour (0-23)                                                    |
| minute      | minute (0-59)                                                  |
| second      | second (0-59)                                                  |

Answer 12

这是一个使用 Thread 的快速而肮脏的非阻塞循环：

#!/usr/bin/env python3
import threading,time

def worker():
    print(time.time())
    time.sleep(5)
    t = threading.Thread(target=worker)
    t.start()


threads = []
t = threading.Thread(target=worker)
threads.append(t)
t.start()
time.sleep(7)
print("Hello World")

没有什么特别的，worker 会延迟创建一个自己的新线程。可能不是最有效的，但足够简单。如果您需要更复杂的解决方案，northtree 的答案将是您的最佳选择。

基于这个，我们可以做同样的事情，只需使用Timer：

#!/usr/bin/env python3
import threading,time

def hello():
    t = threading.Timer(10.0, hello)
    t.start()
    print( "hello, world",time.time() )

t = threading.Timer(10.0, hello)
t.start()
time.sleep(12)
print("Oh,hai",time.time())
time.sleep(4)
print("How's it going?",time.time())

Answer 13

看我的样本

import sched, time

def myTask(m,n):
  print n+' '+m

def periodic_queue(interval,func,args=(),priority=1):
  s = sched.scheduler(time.time, time.sleep)
  periodic_task(s,interval,func,args,priority)
  s.run()

def periodic_task(scheduler,interval,func,args,priority):
  func(*args)
  scheduler.enter(interval,priority,periodic_task,
                   (scheduler,interval,func,args,priority))

periodic_queue(1,myTask,('world','hello'))

Answer 14

有一个新包，名为 ischedule。对于这种情况，解决方案可能如下：

from ischedule import schedule, run_loop
from datetime import timedelta


def query_rate_limit():
    print("query_rate_limit")

schedule(query_rate_limit, interval=60)
run_loop(return_after=timedelta(hours=1))

一切都在主线程上运行，并且 run_loop 内没有忙等待。启动时间非常精确，通常在指定时间的几分之一毫秒内。