不能对保存的表单使用 celery 延迟：对象不是 JSON 可序列化的

Question

使用 Django 1.8，我想在视图中保存表单后触发延迟的芹菜函数

def new_topic(request, forum_id):
    form = TopicForm()
    uid = request.user.id
    if request.method == 'POST':
        tform = TopicForm(request.POST)
        if tform.is_valid():
            topic = tform.save(commit=False)
            topic.title = clean_title(tform.cleaned_data['title'])
            topic.description = clean_desc(tform.cleaned_data['description'])
            topic.save()
            notify_new_topic.delay( uid, topic) #<--problem here
            #rest of the views

但我明白了

EncodeError at /add/topic/
<Topic: Topic object> is not JSON serializable

如果我从 celery 任务中删除 delay，我不会收到任何错误。

任务是：

@shared_task
def notify_new_topic(flwd_id, topic):
    title = topic.title
    link = topic.slug

    flwd= cached_user(flwd_id) #User.objects.get(id = flwd_id)
    print 'flwd is', flwd.username
    flwr_ids = FollowUser.objects.filter(followed=flwd).values('follower_id')
    flwrs = User.objects.filter(id__in= flwr_ids).values('id', 'username','email') 

    for f in flwrs:
        print 'flwr username:',  f['username']
        if notify_flwdp_applies(int(f['id'])):
            print 'notify flwdp applies'
            make_alerts_new_topic(flwd_id, f['id'], topic)
            print 'back from make_alerts_new_topic'

我想知道我该如何 debug/fix 这个？

Answer 1

任务的参数应该是可序列化的（即字符串、整数等）。要修复错误，您可以将 topic_id 作为参数传递并在任务方法中获取主题对象：

notify_new_topic.delay( uid, topic.id)

@shared_task
def notify_new_topic(flwd_id, topic_id):
    topic = Topic.objects.get(pk=topic_id)
    title = topic.title
    link = topic.slug

    flwd= cached_user(flwd_id) #User.objects.get(id = flwd_id)
    print 'flwd is', flwd.username
    flwr_ids = FollowUser.objects.filter(followed=flwd).values('follower_id')
    flwrs = User.objects.filter(id__in= flwr_ids).values('id', 'username','email') 

    for f in flwrs:
        print 'flwr username:',  f['username']
        if notify_flwdp_applies(int(f['id'])):
            print 'notify flwdp applies'
            make_alerts_new_topic(flwd_id, f['id'], topic)
            print 'back from make_alerts_new_topic'

Answer 2

既然已经提供了解决方案，我将尝试解释为什么我们不能将不可序列化的对象传递给celery tasks.

为什么我们需要将可序列化对象传递给 celery 任务？

对于 celery，我们使用 消息代理 （例如 Redis 或 RabbitMQ）。假设我们使用Redis。当调用 celery task 时，参数会传递给 Redis 以便 broker 可以读取它们。为此，Redis.

应支持这些参数的数据类型

解决方法

假设您要将 python dictionary 作为参数传递给 celery 任务，将这些值添加到 celery 配置中：

task_serializer = "json"  
result_serializer = "json"
accept_content = ["json"]

或者你可能想做

celery.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"]
)

其他情况，将上面的json替换为pickle、xml等

典型的基于文本的序列化格式是 csv、json、xml、yaml、toml 等。基于二进制的格式是 protobuf 和 avro。 Python 也有几个包，如 pickle、numpy 和 pandas 支持将自定义对象序列化为 byte 格式。您还可以制作自定义序列化程序。

这些配置有什么作用？

指示芹菜首先序列化python对象，然后将它们传递给消息代理。
从消息代理反序列化对象，然后将它们提供给celery worker。

参考资料

不能对保存的表单使用 celery 延迟：对象不是 JSON 可序列化的

Can not use celery delay for saved form: object is not JSON serializable

django

django-celery

为什么我们需要将可序列化对象传递给 celery 任务？

解决方法

这些配置有什么作用？