python之Celery/RabbitMQ/Django 没有运行任务

我希望有人能帮助我，因为我查看了 Stack Overflow 却找不到解决我问题的方法。我正在运行一个 Django 项目并安装了 Supervisor、RabbitMQ 和 Celery。 RabbitMQ 启动并运行，Supervisor 确保我的 celerybeat 正在运行，但是，虽然它记录节拍已启动并每 5 分钟发送一次任务(见下文)，但任务从未真正执行:

我的主管程序配置文件:

[program:nrv_twitter] 
; Set full path to celery program if using virtualenv 
command=/Users/tsantor/.virtualenvs/nrv_env/bin/celery beat -A app --loglevel=INFO --pidfile=/tmp/nrv-celerybeat.pid --schedule=/tmp/nrv-celerybeat-schedule 
 
; Project dir 
directory=/Users/tsantor/Projects/NRV/nrv 
 
; Logs 
stdout_logfile=/Users/tsantor/Projects/NRV/nrv/logs/celerybeat_twitter.log 
redirect_stderr=true 
 
autorestart=true 
autostart=true 
startsecs=10 
user=tsantor 
 
; if rabbitmq is supervised, set its priority higher so it starts first 
priority=999 

这是上面程序的日志输出:

[2014-12-16 20:29:42,293: INFO/MainProcess] beat: Starting... 
[2014-12-16 20:34:08,161: INFO/MainProcess] Scheduler: Sending due task gettweets-every-5-mins (twitter.tasks.get_tweets) 
[2014-12-16 20:39:08,186: INFO/MainProcess] Scheduler: Sending due task gettweets-every-5-mins (twitter.tasks.get_tweets) 
[2014-12-16 20:44:08,204: INFO/MainProcess] Scheduler: Sending due task gettweets-every-5-mins (twitter.tasks.get_tweets) 
[2014-12-16 20:49:08,205: INFO/MainProcess] Scheduler: Sending due task gettweets-every-5-mins (twitter.tasks.get_tweets) 
[2014-12-16 20:54:08,223: INFO/MainProcess] Scheduler: Sending due task gettweets-every-5-mins (twitter.tasks.get_tweets) 

这是我的 celery.py 设置文件:

from datetime import timedelta 
 
BROKER_URL = 'amqp://guest:guest@localhost//' 
 
CELERY_DISABLE_RATE_LIMITS = True 
 
CELERYBEAT_SCHEDULE = { 
    'gettweets-every-5-mins': { 
        'task': 'twitter.tasks.get_tweets', 
        'schedule': timedelta(seconds=300) # 300 = every 5 minutes 
    }, 
} 

这是我的celeryapp.py:

from __future__ import absolute_import 
import os 
from django.conf import settings 
from celery import Celery 
 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'app.settings') 
 
app = Celery('app') 
app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 

这是我的twitter/tasks.py:

from __future__ import absolute_import 
import logging 
from celery import shared_task 
from twitter.views import IngestTweets 
 
log = logging.getLogger('custom.log') 
 
@shared_task 
def get_tweets(): 
    """ 
    Get tweets and save them to the DB 
    """ 
    instance = IngestTweets() 
    IngestTweets.get_new_tweets(instance) 
 
    log.info('Successfully ingested tweets via celery task') 
    return True 

get_tweets 方法永远不会执行，但我知道它可以正常工作，因为我可以手动执行 get_tweets 并且它工作正常。

我花了两天时间想弄清楚为什么它发送了应有的任务，但不执行它们？任何帮助是极大的赞赏。提前致谢。

请您参考如下方法:

user2097159 感谢您为我指明了正确的方向，我不知道我还必须使用 supervisor 运行一个 worker。我认为它要么是一个 worker 要么是一个节拍，但现在我明白我必须有一个 worker 来处理任务和一个节拍来定期触发任务。

下面是主管缺少的 worker 配置:

[program:nrv_celery_worker] 
; Worker 
command=/Users/tsantor/.virtualenvs/nrv_env/bin/celery worker -A app --loglevel=INFO 
 
; Project dir 
directory=/Users/tsantor/Projects/NRV/nrv 
 
; Logs 
stdout_logfile=/Users/tsantor/Projects/NRV/nrv/logs/celery_worker.log 
redirect_stderr=true 
 
autostart=true 
autorestart=true 
startsecs=10 
user=tsantor 
numprocs=1 
 
; Need to wait for currently executing tasks to finish at shutdown. 
; Increase this if you have very long running tasks. 
stopwaitsecs = 600 
 
; When resorting to send SIGKILL to the program to terminate it 
; send SIGKILL to its whole process group instead, 
; taking care of its children as well. 
killasgroup=true 
 
; if rabbitmq is supervised, set its priority higher 
; so it starts first 
priority=998 

然后我重置了 RabbitMQ 队列。现在我已经通过主管管理了节拍和 worker 程序，一切都按预期进行。希望这可以帮助其他人。