芹菜 + Django 信号

Question

我正在尝试将 Django Signals 的 post_save 函数与 Celery 任务结合使用。将新的 Message 对象保存到数据库后，我想评估该实例是否具有两个属性之一，如果有，则调用 'send_sms_function'，它是 Celery 注册的任务。

任务.py

from my_project.celery import app

@app.task
def send_sms_message(message):
    # Do something

信号.py

from django.db.models.signals import post_save
from django.dispatch import receiver

import rollbar
rollbar.init('234...0932', 'production')

from dispatch.models import Message
from comm.tasks import send_sms_message


@receiver(post_save, sender=Message)
def send_outgoing_messages(sender, instance, **kwargs):

if instance.some_attribute == 'A' or instance.some_attribute == 'B':
    try:
        send_sms_message.delay(instance)
    except:
         rollbar.report_exc_info()
else:
    pass

我正在通过运行 Celery 工作器在本地测试它。当我在 Django shell 中调用 Celery 函数时，它按预期工作。但是，当我将 Message 实例保存到数据库时，该函数无法按预期工作：任务队列中没有发布任何内容，也没有看到任何错误消息。

我究竟做错了什么？

Answer 1

这看起来像是序列化和/或您的设置的问题。当 celery 将消息传递给您的代理时，它需要对数据进行某种表示。Celery 序列化您提供给任务的参数，但如果您没有将其配置为与您传递的内容一致（即您的经纪人期望 JSON 不匹配，但您将其发送给一个腌制的 Python 对象），任务可能会失败因为工作人员无法轻松解码您发送的内容。如果您在 shell 中运行该函数（没有调用延迟），它会被同步调用，因此没有序列化或消息传递。

在您的设置中，您应该使用 JSON 序列化（除非您有很好的理由），但如果没有，那么您的酸洗可能有问题。当您运行 celery 以查看有关序列化相关错误的更多信息时，您始终可以增加日志级别以进行调试：

celery -A yourapp worker -l debug

如有疑问，请使用该打印语句/函数来确保您的信号接收器正在运行。如果没有，您可以创建一个AppConfig类，在它的ready方法或其他一些合理的技术中导入您的接收器，以确保您的接收器被注册。

[意见]我建议做这样的事情：

@receiver(post_save, sender=Message)
def send_outgoing_messages(sender, instance, **kwargs):
    enqueue_message.delay(instance.id)

在 yourmodule/tasks.py

@app.task
def enqueue_message(message_id):
    msg = Message.object.get(id=message_id)
    if msg.some_attribute in ('A', 'B'):  # slick or
        send_sms_message.delay(message_id)

你总是可以使用 Celery 的组合技术，但在这里你有一些不会给你的请求/响应周期增加更多复杂性的东西。[/观点]

Answer 2

可能是你的表达方式if instance.some_attribute == 'A' or 'B'有问题。

你的意思可能是：

if instance.some_attribute == 'A' or instance.some_attribute == 'B'

或者，我会怎么写：

if instance.some_attribute in ('A', 'B')