Celery消息队列----配置定时任务

时间：2021-11-04 07:46:17

介绍

celery 定时器是一个调度器（scheduler）；它会定时地开启（kicks off）任务，然后由集群中可用的工人（worker）来执行。

定时任务记录（entries）默认 从 beat_schedule 设置中获取，但自定义存储也可以使用，如把记录存储到SQL数据库中。

要确保同一时间一份时间表上只有一个调度器在运行，否则会因为重复发送任务而结束。使用集中途径意味着定时任务不用必须同步，并且服务无需用锁操控。

记录

为了定时调用任务，你必须添加记录到打点列表中：

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# 每10秒调用 test('hello') .
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

# 每30秒调用 test('world') 
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

# 每周一上午7:30执行
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

用on_after_configure处理器进行这些设置意味着当使用test.s()时我们不会在模块层面运行app 。

add_periodic_task() 函数在幕后会添加记录到beat_schedule设定，同样的设定可以用来手动设置定时任务：

例子: 每30秒运行 tasks.add .

app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16)
    },
}
app.conf.timezone = 'UTC'

一般会使用配置文件进行配置，如下
celeryconfig.py:

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True
beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16)
    },
}

程序里使用：

app.config_from_object('celeryconfig')
注意
如果你的参数元组里只有一个项目，只用一个逗号就可以了，不要圆括号。

时间表使用时间差意味着每30秒间隔会发送任务（第一个任务在celery定时器开启后30秒发送，然后上每次距一次运行后30秒发送一次）

可使用的属性

  • task

    要执行的任务名字

  • schedule

    执行的频率

    可以是整数秒数，时间差，或者一个周期（ crontab）。你也可以自 定义你的时间表类型，通过扩展schedule接口。

  • args

    位置参数 (list 或 tuple).

  • kwargs

    键值参数 (dict).

  • options

    执行选项 (dict).

    这可以是任何被apply_async()支持的参数与—-exchange, routing_key, expires,等。

  • relative

    如果 relative 是 true ，时间表“由时钟时间”安排，意味着 频率近似到最近的秒，分钟，小时或天，这取决于时间差中的时间间隔。
    默认relative是false，频率不会近似，会相对于celery的启动时间。

    Crontab 表达式语法非常灵活。

例子 含义
crontab() 每分钟执行
crontab(minute=0, hour=0) 每天午夜执行
crontab(minute=0, hour=’*/3’) 每三个小时执行: 午夜, 3am, 6am, 9am, 正午, 3pm, 6pm, 9pm.
crontab(minute=0,hour=’0,3,6,9,12,15,18,21’) 同上
crontab(minute=’*/15’) 每15分钟执行
crontab(day_of_week=’sunday’) 星期日每分钟
crontab(minute=’‘,hour=’‘, day_of_week=’sun’) 同上
crontab(minute=’*/10’,hour=’3,17,22’, day_of_week=’thu,fri’) 每10分钟执行，仅限于周六日3-4 am, 5-6 pm, and 10-11 pm
crontab(minute=0, hour=’/2,/3’) 偶数小时或者能被3整除的小时数执行
crontab(minute=0, hour=’*/5’) 被5整除的小时数，如3pm
crontab(minute=0, hour=’*/3,8-17’) 8am-5pm能被3整除的
crontab(0, 0, day_of_month=’2’) 每月第2天
crontab(0, 0,day_of_month=’2-30/3’) 每偶数天
crontab(0, 0,day_of_month=’1-7,15-21’) 每月1和3周
crontab(0, 0, day_of_month=’11’,month_of_year=’5’) 每年5月11日
crontab(0, 0,month_of_year=’*/3’) 每个季度第1月

开启调度

开启celery定时服务:

$ celery -A proj beat

你也可以把定时器嵌入到工人（worker）中，通过启用workers -B选项，如果你永远不会运行超过一个工人节点这就会很方便。但这不太常见，不推荐在生产环境这样使用：

$ celery -A proj worker -B

定时器需要在本地数据库文件(默认名为 celerybeat-schedule )存储任务上次运行时间，所以它需要在当前目录中写权限。或者你也可以给这个文件指定一个位置：

$ celery -A proj beat -s /home/celery/var/run/celerybeat-schedule
标签：

相关文章