不能对保存的表单使用 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

如果我从 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'

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

Answer 1

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

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

对于 celery，我们使用 消息代理（如 Redis 或 RabbitMQ）。假设我们使用 Redis。当调用 celery 任务 时，会将参数传递给 Redis，以便代理可以读取它们。为了实现这一点，这些参数的数据类型应该由 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 格式。您还可以制作自定义序列化程序。

这些配置有什么作用？

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

参考文献

• Tutorials Point: Object-Oriented Python - Object Serialization
• Issue on Github: Type Object not JSON serializable
• towards data science: What, Why and How of (De)Serialization in Python

Answer 2

Task 的参数应该是可序列化的（即 string、int 等）。要修复错误，您可以传递 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'