如何使用每小時查詢數據庫的Celerybeat和Flask設置定期任務?安裝芹菜週期性任務
的環境是這樣的:
/
|-app
|-__init__.py
|-jobs
|-task.py
|-celery-beat.sh
|-celery-worker.sh
|-manage.py
我現在有一個名爲run_query()
查詢功能位於task.py
我想要調度程序踢一旦應用程序啓動,所以我有下面幾行我/app/__init__.py
文件夾:
celery = Celery()
@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
sender.add_periodic_task(1, app.jobs.task.run_query())
(爲簡單起見,我已經設定,使如果它運行,它會每分鐘運行一次。沒有這樣的運氣呢。)
當我啓動celery-worker.sh
它確認我的功能在[tasks]
標題下。但計劃的函數從不運行。我可以手動迫使功能通過發出運行在命令提示以下:
>> from app.jobs import task
>> task.run_query.delay()
EDIT:添加celerybeat.sh
作爲後續:如果數據庫是通過燒瓶上下文中訪問,在我的異步函數調用期間,創建一個新的上下文來訪問數據庫是明智的嗎?使用現有的燒瓶環境?或者完全忘記上下文,只是啓動到數據庫的連接?我擔心的是,如果我只是啓動一個新的連接,它可能會干擾現有的上下文的連接?
您還需要運行調度服務。 http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#starting-the-scheduler。 **芹菜擊敗**是一個調度程序;它定期開始執行任務,然後由集羣中可用的工作節點執行。 –