Celeryタイミングタスクの簡単な例

需要:5分ごとにブログリストを検索
手順:
redisサーバを開く
ローカルプロジェクトサーバを開く
 
新しいtasksファイルコードは次のとおりです.

# coding:utf-8
import requests
from celery import Celery


app = Celery('tasks', broker='redis://127.0.0.1:6379/0')

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    #       
    sender.add_periodic_task(300, test.s('http://127.0.0.1:8000/blog/add/'), name='add every 5minites')

@app.task
def test(url):
    print('      ,,,,,    ')
    requests.get(url)

注意celeryバージョンは4.0以上のバージョンが必要です
tasksと同級ディレクトリに移動

celery        ：
case_manage>celery -A tasks beat

celery     ：
celery -A tasks worker --loglevel=info --pool=solo

タイマーが正常に停止していない場合、タイマーを再起動したときに次のエラーが発生した場合:

celery beat v4.1.1 (latentcall) is starting.
ERROR: Pidfile (celerybeat.pid) already exists.
Seems we're already running? (pid: 35544)

tasksの兄弟ディレクトリの下のcelerybeatを削除するだけです.pidファイルを再起動すればいい!