未执行的任务（Django + Heroku + Celery + RabbitMQ）

Question

我是第一次使用 RabbitMQ，我一定对一些简单的配置设置有误解。 请注意，我现在在本地运行应用程序时遇到了这个问题；我还没有尝试通过 Heroku 投入生产。

对于这个应用程序，我想每隔 20 秒在数据库中查找一些未发送的消息，并通过 Twilio 发送它们。如果我在下面的示例中遗漏了一些相关代码，请提前道歉。我已遵循所有 Celery 设置/配置说明。这是我目前的设置：

BROKER_URL = 'amqp://VflhnMEP:8wGLOrNBP.........Bhshs'  # Truncated URL string

from datetime import timedelta
CELERYBEAT_SCHEDULE = {
    'send_queued_messages_every_20_seconds': {
        'task': 'comm.tasks.send_queued_messages',
        'schedule': timedelta(seconds=20),
        # 'schedule': crontab(seconds='*/20')
        },
    }
CELERY_TIMEZONE = 'UTC'

我很确定这些任务正在 RabbitMQ 中完成；这是我可以看到的所有累积消息的破折号：

“send_queued_messages”函数应该每 20 秒调用一次。

comm/tasks.py

导入日期时间 从 celery.decorators 导入周期性任务

from comm.utils import get_user_mobile_number
from comm.api import get_twilio_connection, send_message
from dispatch.models import Message

@periodic_task
def send_queued_messages(run_every=datetime.timedelta(seconds=20)):
    unsent_messages = Message.objects.filter(sent_success=False)
    connection = get_twilio_connection()

    for message in unsent_messages:
        mobile_number = get_user_mobile_number(message=message)
        try:
            send_message(
                connection=connection,
                mobile_number=mobile_number,
                message=message.raw_text
                )
            message.sent_success=True
            message.save()
        except BaseException as e:
            raise e
            pass

我很确定我在 RabbitMQ 或我的 Heroku 项目设置中配置错误，但我不确定如何继续进行故障排除。当我运行“celery -A myproject beat”时一切似乎都很顺利。

(venv)josephs-mbp:myproject josephfusaro$ celery -A myproject beat
celery beat v3.1.18 (Cipater) is starting.
__    -    ... __   -        _
Configuration ->
    . broker -> amqp://VflhnMEP:**@happ...Bhshs
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%INFO
    . maxinterval -> now (0s)
[2015-05-27 03:01:53,810: INFO/MainProcess] beat: Starting...
[2015-05-27 03:02:13,941: INFO/MainProcess] Scheduler: Sending due task send_queued_messages_every_20_seconds (comm.tasks.send_queued_messages)
[2015-05-27 03:02:34,036: INFO/MainProcess] Scheduler: Sending due task send_queued_messages_every_20_seconds (comm.tasks.send_queued_messages)

那么，为什么在没有 Celery 参与的情况下，任务不能像它们那样执行*？

我的档案：

web: gunicorn myproject.wsgi --log-file - worker: celery -A myproject beat

*我已经确认我的代码在没有涉及 Celery 的情况下按预期执行！

Answer 1

特别感谢@MauroRocco 将我推向正确的方向。我错过的部分在本教程中得到了最好的解释：https://www.rabbitmq.com/tutorials/tutorial-one-python.html

注意：我需要修改教程中的一些代码以使用 URLParameters，传入我的设置文件中定义的资源 URL。

send.py 和 receive.py 中唯一的一行是：

connection = pika.BlockingConnection(pika.URLParameters(BROKER_URL))

当然我们需要从 settings.py 中导入 BROKER_URL 变量

from settings import BROKER_URL

settings.py

BROKER_URL = 'amqp://VflhnMEP:8wGLOrNBP...4.bigwig.lshift.net:10791/sdklsfssd'

send.py

import pika
from settings import BROKER_URL

connection = pika.BlockingConnection(pika.URLParameters(BROKER_URL))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')
print " [x] Sent 'Hello World!'"
connection.close()

receive.py

import pika
from settings import BROKER_URL

connection = pika.BlockingConnection(pika.URLParameters(BROKER_URL))
channel = connection.channel()

channel.queue_declare(queue='hello')

print ' [*] Waiting for messages. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)

channel.start_consuming()