无法使用 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

任务的参数应该是可序列化的（即字符串、整数等）。要修复错误，您可以传递

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 支持。

解决方法

假设您想将

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工作者。

参考文献

• 教程要点：面向对象的Python - 对象序列化
• Github 上的问题：类型对象不可 JSON 序列化
• 迈向数据科学：Python 中（反）序列化的内容、原因和方式

Answer 3

更改为picke编码

app.conf.event_serializer = 'pickle' # this event_serializer is optional. 
app.conf.task_serializer = 'pickle'
app.conf.result_serializer = 'pickle'
app.conf.accept_content = ['application/json', 'application/x-python-serialize']