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

Question

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

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

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

任务是：

@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'

我想知道如何调试/修复这个问题？

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 任务。

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

对于 celery，我们使用消息代理（例如Redis或RabbitMQ）。假设我们使用Redis。当调用celery 任务时，参数会传递到Redis，以便代理可以读取它们。为此，这些参数的数据类型应该受到Redis支持。

解决方法

假设您想将 apython 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格式。您还可以制作自定义序列化器。

这些配置有什么作用？

1. 指示 celery 首先序列化 python 对象，然后将它们传递给消息代理。
2. 反序列化来自消息代理的对象，然后将它们提供给celery worker。

参考

• 教程要点：面向对象的 Python - 对象序列化
• Github 上的问题：类型对象不可 JSON 序列化
• 迈向数据科学：Python 中的（反）序列化是什么、为什么以及如何进行