我想使用django_rq
和rq-scheduler
进行离线任务,但是我不确定在哪里调用rq-scheduler
的能力来安排重复任务。目前,我已将我的调度添加到我的应用程序中的tasks.py
模块,并在__init__.py
中导入该模块。不过,一定有更好的方法吧?
提前感谢。
我想使用django_rq
和rq-scheduler
进行离线任务,但是我不确定在哪里调用rq-scheduler
的能力来安排重复任务。目前,我已将我的调度添加到我的应用程序中的tasks.py
模块,并在__init__.py
中导入该模块。不过,一定有更好的方法吧?
提前感谢。
django_rq
中包含的rqscheduler
命令。这里提供了一个示例:https://github.com/rq/rq-scheduler/issues/51#issuecomment-362352497。rqscheduler
管理命令进行了子类化,以便同时调用scheduler.schedule
命令。我曾经遇到过schedule方法运行多次的问题,而这个方法解决了这个问题。感谢@Pete! - davidscolgan我发现最好的运行位置是从你的AppConfig
中的apps.py
。
def ready(self):
scheduler = django_rq.get_scheduler('default')
# Delete any existing jobs in the scheduler when the app starts up
for job in scheduler.get_jobs():
job.delete()
# Have 'mytask' run every 5 minutes
scheduler.schedule(datetime.utcnow(), 'mytask', interval=60*5)
我在我的项目应用程序(基于Django)__init__
模块中添加了调度功能,但使用小函数进行包装,以防止重复排队作业。调度策略可能取决于您的特定需求(例如,您可能需要对作业参数进行额外检查)。
以下是适合我需求并且可行的代码:
import django_rq
from collections import defaultdict
import tasks
scheduler = django_rq.get_scheduler('default')
jobs = scheduler.get_jobs()
functions = defaultdict(lambda: list())
map(lambda x: functions[x.func].append(x.meta.get('interval')), jobs)
now = datetime.datetime.now()
def schedule_once(func, interval):
"""
Schedule job once or reschedule when interval changes
"""
if not func in functions or not interval in functions[func]\
or len(functions[func])>1:
# clear all scheduled jobs for this function
map(scheduler.cancel, filter(lambda x: x.func==func, jobs))
# schedule with new interval
scheduler.schedule(now+datetime.timedelta(seconds=interval), func,
interval=interval)
schedule_once(tasks.some_task_a, interval=60*5)
schedule_once(tasks.some_task_b, interval=120)
def init_scheduler():
# paste here initialization code
init_scheduler()
您应该使用Django命令来运行计划任务 https://docs.djangoproject.com/en/3.2/howto/custom-management-commands/
像这样
class Command(BaseCommand):
def handle(self, *args, **options):
scheduler = django_rq.get_scheduler('crontab_job')
for job in scheduler.get_jobs():
scheduler.cancel(job)
# 定时任务例子1
scheduler.cron(
"*/3 * * * *", # 每周一零点零时零分执行 0 0 * * 0 测试可以使用 */3 * * * * 每3分钟执行一次
func=gong_an_job, # Function to be queued
kwargs={'msg': '我是王龙飞1,我喜欢修仙', 'number': 1}, # Keyword arguments passed into function when executed
repeat=None, # Repeat this number of times (None means repeat forever)
queue_name='crontab_job', # In which queue the job should be put in
use_local_timezone=False # Interpret hours in the local timezone
)
# 定时任务例子2
scheduler.cron(
"*/5 * * * *", # 每周一零点零时零分执行 0 0 * * 0 测试可以使用 */3 * * * * 每3分钟执行一次
func=gong_an_job, # Function to be queued
kwargs={'msg': '我是王龙飞222222,我喜欢修仙', 'number': 22222}, # Keyword arguments passed into function when executed
repeat=None, # Repeat this number of times (None means repeat forever)
queue_name='crontab_job', # In which queue the job should be put in
use_local_timezone=False # Interpret hours in the local timezone
)
python manage.py rq_crontab_job
python manage.py rqscheduler --queue crontab_job
python manage.py rqworker crontab_job
我认为第一个答案很好,但在多进程环境中可能会出现一些问题,你只需要运行一次来创建crontab任务!
rq:scheduler:scheduled_jobs
键,然后再添加它们。但请记住避免在工作进程中添加/删除作业 - 我不知道如何实现这一点。 - marcinn