如何在Python中获得类似Cron的调度程序？-Java 学习之路

258

我正在寻找一个Python库，它将提供 at 和 cron 之类的功能 .

我非常喜欢使用纯Python解决方案，而不是依赖于盒子上安装的工具;这样我就可以在没有cron的机器上运行 .

对于那些不熟悉 cron 的人：您可以根据以下表达式安排任务：

0 2 * * 7 /usr/bin/run-backup # run the backups at 0200 on Every Sunday
 0 9-17/2 * * 1-5 /usr/bin/purge-temps # run the purge temps command, every 2 hours between 9am and 5pm on Mondays to Fridays.

cron时间表达式语法不那么重要，但我希望能有这种灵活性 .

如果没有任何东西可以为我开箱即用，那么任何关于构建模块的建议都会感激不尽 .

Edit 我对启动进程不感兴趣，只是"jobs"也是用Python编写的 - python函数 . 必要时我认为这将是一个不同的主题，但不是在一个不同的过程中 .

为此，我正在寻找cron时间表达式的表现力，但在Python中 .

Cron已经存在多年了，但我试图尽可能地便携 . 我不能依赖它的存在 .

21 回答

如果您正在寻找轻量级结账schedule：

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)

披露：我是该图书馆的作者 .

回复于 2024-04-29T16:17:09+08:00

您可以使用普通的Python参数传递语法来指定您的crontab . 例如，假设我们定义一个Event类，如下所示：

from datetime import datetime, timedelta
import time

# Some utility classes / functions first
class AllMatch(set):
    """Universal set - match everything"""
    def __contains__(self, item): return True

allMatch = AllMatch()

def conv_to_set(obj):  # Allow single integer to be provided
    if isinstance(obj, (int,long)):
        return set([obj])  # Single item
    if not isinstance(obj, set):
        obj = set(obj)
    return obj

# The actual Event class
class Event(object):
    def __init__(self, action, min=allMatch, hour=allMatch, 
                       day=allMatch, month=allMatch, dow=allMatch, 
                       args=(), kwargs={}):
        self.mins = conv_to_set(min)
        self.hours= conv_to_set(hour)
        self.days = conv_to_set(day)
        self.months = conv_to_set(month)
        self.dow = conv_to_set(dow)
        self.action = action
        self.args = args
        self.kwargs = kwargs

    def matchtime(self, t):
        """Return True if this event should trigger at the specified datetime"""
        return ((t.minute     in self.mins) and
                (t.hour       in self.hours) and
                (t.day        in self.days) and
                (t.month      in self.months) and
                (t.weekday()  in self.dow))

    def check(self, t):
        if self.matchtime(t):
            self.action(*self.args, **self.kwargs)

（注意：未经过彻底测试）

然后你的CronTab可以用普通的python语法指定为：

c = CronTab(
  Event(perform_backup, 0, 2, dow=6 ),
  Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)

通过这种方式，您可以获得Python的参数机制的全部功能（混合位置和关键字args，并且可以使用符号名称来表示数周和数月的名称）

CronTab类将被定义为以微小的增量进行简单的睡眠，并在每个事件上调用check（） . （可能会有一些细微之处，夏令时/时区要警惕） . 这是一个快速实现：

class CronTab(object):
    def __init__(self, *events):
        self.events = events

    def run(self):
        t=datetime(*datetime.now().timetuple()[:5])
        while 1:
            for e in self.events:
                e.check(t)

            t += timedelta(minutes=1)
            while datetime.now() < t:
                time.sleep((t - datetime.now()).seconds)

需要注意的一点是：Python的工作日/月是零索引（与cron不同），并且该范围排除了最后一个元素，因此像“1-5”这样的语法变为范围（0,5） - 即[0,1,2， 3,4] . 如果您更喜欢cron语法，那么解析它应该不会太困难 .

回复于 2024-04-29T16:17:09+08:00

1

也许只有在提出问题之后才会出现这种情况;我以为我只是为了完整起见而提到它：https://apscheduler.readthedocs.org/en/latest/

回复于 2024-04-29T16:17:09+08:00
0

查看Celery，他们有像cron这样的定期任务 .

回复于 2024-04-29T16:17:09+08:00
27

"... Crontab module for read and writing crontab files and accessing the system cron automatically and simply using a direct API. ..."

http://pypi.python.org/pypi/python-crontab

以及APScheduler，一个python包 . 已经编写和调试 .

http://packages.python.org/APScheduler/cronschedule.html

回复于 2024-04-29T16:17:09+08:00
19

在我的搜索中，有一件事我可能就是你正在寻找的那种东西.1111886_ s sched module .

回复于 2024-04-29T16:17:09+08:00

或多或少与上面相同，但使用gevent并发:)

"""Gevent based crontab implementation"""

from datetime import datetime, timedelta
import gevent

# Some utility classes / functions first
def conv_to_set(obj):
    """Converts to set allowing single integer to be provided"""

    if isinstance(obj, (int, long)):
        return set([obj])  # Single item
    if not isinstance(obj, set):
        obj = set(obj)
    return obj

class AllMatch(set):
    """Universal set - match everything"""
    def __contains__(self, item): 
        return True

allMatch = AllMatch()

class Event(object):
    """The Actual Event Class"""

    def __init__(self, action, minute=allMatch, hour=allMatch, 
                       day=allMatch, month=allMatch, daysofweek=allMatch, 
                       args=(), kwargs={}):
        self.mins = conv_to_set(minute)
        self.hours = conv_to_set(hour)
        self.days = conv_to_set(day)
        self.months = conv_to_set(month)
        self.daysofweek = conv_to_set(daysofweek)
        self.action = action
        self.args = args
        self.kwargs = kwargs

    def matchtime(self, t1):
        """Return True if this event should trigger at the specified datetime"""
        return ((t1.minute     in self.mins) and
                (t1.hour       in self.hours) and
                (t1.day        in self.days) and
                (t1.month      in self.months) and
                (t1.weekday()  in self.daysofweek))

    def check(self, t):
        """Check and run action if needed"""

        if self.matchtime(t):
            self.action(*self.args, **self.kwargs)

class CronTab(object):
    """The crontab implementation"""

    def __init__(self, *events):
        self.events = events

    def _check(self):
        """Check all events in separate greenlets"""

        t1 = datetime(*datetime.now().timetuple()[:5])
        for event in self.events:
            gevent.spawn(event.check, t1)

        t1 += timedelta(minutes=1)
        s1 = (t1 - datetime.now()).seconds + 1
        print "Checking again in %s seconds" % s1
        job = gevent.spawn_later(s1, self._check)

    def run(self):
        """Run the cron forever"""

        self._check()
        while True:
            gevent.sleep(60)

import os 
def test_task():
    """Just an example that sends a bell and asd to all terminals"""

    os.system('echo asd | wall')  

cron = CronTab(
  Event(test_task, 22, 1 ),
  Event(test_task, 0, range(9,18,2), daysofweek=range(0,5)),
)
cron.run()

回复于 2024-04-29T16:17:09+08:00

0

TurboGears附带基于Kronos的预定任务能力

我从来没有直接使用过Kronos，但TG中的调度具有一套不错的功能并且非常可靠 .

回复于 2024-04-29T16:17:09+08:00

列出的解决方案都没有尝试解析复杂的cron调度字符串 . 所以，这是我的版本，使用croniter . 基本要点：

schedule = "*/5 * * * *" # Run every five minutes

nextRunTime = getNextCronRunTime(schedule)
while True:
     roundedDownTime = roundDownTime()
     if (roundedDownTime == nextRunTime):
         ####################################
         ### Do your periodic thing here. ###
         ####################################
         nextRunTime = getNextCronRunTime(schedule)
     elif (roundedDownTime > nextRunTime):
         # We missed an execution. Error. Re initialize.
         nextRunTime = getNextCronRunTime(schedule)
     sleepTillTopOfNextMinute()

帮助程序：

from croniter import croniter
from datetime import datetime, timedelta

# Round time down to the top of the previous minute
def roundDownTime(dt=None, dateDelta=timedelta(minutes=1)):
    roundTo = dateDelta.total_seconds()
    if dt == None : dt = datetime.now()
    seconds = (dt - dt.min).seconds
    rounding = (seconds+roundTo/2) // roundTo * roundTo
    return dt + timedelta(0,rounding-seconds,-dt.microsecond)

# Get next run time from now, based on schedule specified by cron string
def getNextCronRunTime(schedule):
    return croniter(schedule, datetime.now()).get_next(datetime)

# Sleep till the top of the next minute
def sleepTillTopOfNextMinute():
    t = datetime.utcnow()
    sleeptime = 60 - (t.second + t.microsecond/1000000.0)
    time.sleep(sleeptime)

回复于 2024-04-29T16:17:09+08:00

我修改了脚本 .

易于使用：

cron = Cron()
cron.add('* * * * *'   , minute_task) # every minute
cron.add('33 * * * *'  , day_task)    # every hour
cron.add('34 18 * * *' , day_task)    # every day
cron.run()

尝试在第一分钟开始执行任务 .

Code on Github

回复于 2024-04-29T16:17:09+08:00

我对CronTab class run method suggested by Brian有一个小修正 .

时间已经过了一秒钟，在每分钟结束时导致一秒钟的硬循环 .

class CronTab(object):
    def __init__(self, *events):
        self.events = events

    def run(self):
        t=datetime(*datetime.now().timetuple()[:5])
        while 1:
            for e in self.events:
                e.check(t)

            t += timedelta(minutes=1)
            n = datetime.now()
            while n < t:
                s = (t - n).seconds + 1
                time.sleep(s)
                n = datetime.now()

回复于 2024-04-29T16:17:09+08:00

6

查看luigi（https://github.com/spotify/luigi） . 它是用python编写的，有一个很好的Web UI来监控任务 . 它还有一个依赖图 . 对于你需要的东西可能有点过分，但它可能会成功 .

回复于 2024-04-29T16:17:09+08:00
1

你想要的不是't a 1116899 way to do this because some other process would have to launch python in order to run your solution. Every platform will have one or twenty different ways to launch processes and monitor their progress. On unix platforms, cron is the old standard. On Mac OS X there is also launchd, which combines cron-like launching with watchdog functionality that can keep your process alive if that' . python运行后，您可以使用sched module来安排任务 .

回复于 2024-04-29T16:17:09+08:00
10

以防万一，如果你使用的是Windows，那就有一个pycron . 看看http://sourceforge.net/projects/pycron/ . 对于linux，我会选择cron或sched .

回复于 2024-04-29T16:17:09+08:00
7
Brian's solution工作得很好 . 但是，正如其他人所指出的那样，运行代码中存在一个微妙的错误 . 我发现它的需求过于复杂 .

这是我运行代码的简单而实用的替代方案，以防任何人需要它：
```
def run(self):
    while 1:
        t = datetime.now()
        for e in self.events:
            e.check(t)

        time.sleep(60 - t.second - t.microsecond / 1000000.0)
```
回复于 2024-04-29T16:17:09+08:00

另一个简单的解决方案是：

from aqcron import At
from time import sleep
from datetime import datetime

# Event scheduling
event_1 = At( second=5 )
event_2 = At( second=[0,20,40] )

while True:
    now = datetime.now()

    # Event check
    if now in event_1: print "event_1"
    if now in event_2: print "event_2"

    sleep(1)

而aqcron.At类是：

# aqcron.py

class At(object):
    def __init__(self, year=None,    month=None,
                 day=None,     weekday=None,
                 hour=None,    minute=None,
                 second=None):
        loc = locals()
        loc.pop("self")
        self.at = dict((k, v) for k, v in loc.iteritems() if v != None)

    def __contains__(self, now):
        for k in self.at.keys():
            try:
                if not getattr(now, k) in self.at[k]: return False
            except TypeError:
                if self.at[k] != getattr(now, k): return False
        return True

回复于 2024-04-29T16:17:09+08:00

391

如果您正在寻找分布式调度程序，您可以查看https://github.com/sherinkurian/mani - 它确实需要redis，但可能不是您想要的 . （请注意，我是作者）这是为了通过在多个节点上运行时钟来确保容错 .

回复于 2024-04-29T16:17:09+08:00
16

一世不知道这样的事情是否已经存在 . 用时间，日期时间和/或日历模块编写自己的代码很容易，参见http://docs.python.org/library/time.html

python解决方案的唯一问题是你的工作需要一直运行，并且可能在重启后自动"resurrected"，你需要依赖系统相关的解决方案 .

回复于 2024-04-29T16:17:09+08:00
3

我采用了Brian的解决方案，进行了一些更改，添加了标准crontab文件解析器的开头，并将其放在https://bitbucket.org/dbenamy/devcron .

回复于 2024-04-29T16:17:09+08:00
41

你可以查看PiCloud的[1] Crons [2]，但请注意你的工作不会在你自己的机器上运行 . 如果您每月使用超过20小时的计算时间，那么它也是您需要支付的服务 .

[1] http://www.picloud.com

[2] http://docs.picloud.com/cron.html

回复于 2024-04-29T16:17:09+08:00
59

如果要运行的脚本是基于Web的，则可以考虑使用第三方服务（例如crono）以编程方式设置作业 .

回复于 2024-04-29T16:17:09+08:00

如何在Python中获得类似Cron的调度程序？

21 回答

相关问题