在没有 django 的情况下使用 python 在 celery 中执行周期性任务答案

【问题标题】：periodic task in celery using python without django在没有 django 的情况下使用 python 在 celery 中执行周期性任务
【发布时间】：2017-05-12 12:58:46
【问题描述】：

大家好，我是 celery 和 python 的新手。我正在使用 rabbitmq-server 创建一个简单的任务。但我不知道如何在 python 中使用 celery beat 来实现周期性任务。我搜索但每个地方我都会定期使用 django 完成任务。

我将此代码用作tasks.py：从芹菜进口芹菜从时间导入strftime

app = Celery('tasks',broker='pyamqp://guest@localhost//')

@app.task
def show_time():
    return strftime('%Y-%m-%d %H:%M:%S')

run_task.py：

from tasks import show_time
show_time.delay()

感谢您的宝贵时间。

【问题讨论】：

标签： python-2.7 celery-task celerybeat

【解决方案1】：

经过更简洁的搜索，我终于找到了解决方案

from celery import Celery
from kombu import Queue, Exchange



class Config(object):
    CELERY_QUEUES = (
        Queue(
            'try',
            exchange=Exchange('try'),
            routing_key='try',
        ),
    )
celery =Celery('tasks',broker='pyamqp://guest@localhost//')

celery.config_from_object(Config)


celery.conf.beat_schedule = {
    'planner': {
        'task': 'task_planner.some_task',
        'schedule': 5.0,
    },
}


@celery.task(queue='try')
def some_task():
    print('Hooray')

并运行命令： celery -A task_planner worker -l info -B

【讨论】：