Celery能否在单个任务层级指定结果使用的序列化器

可以直接在单个任务定义时指定专属序列化器，配置优先级高于全局celery.conf的设置，不会影响其他任务的默认序列化逻辑。

具体实现方式 #

1. 函数形式定义的任务 #

在@app.task装饰器中直接传入对应序列化器参数即可，针对执行结果序列化场景，传result_serializer参数：

from celery import Celery
app = Celery('your_project')

# 全局配置：默认结果用pickle序列化
app.conf.update(
    result_serializer='pickle',
    # 注意：必须把所有要用到的序列化器加入内容白名单，否则会被安全拦截
    accept_content=['pickle', 'json']
)

# 单独指定这个任务的结果用json序列化，覆盖全局配置
@app.task(result_serializer='json')
def specific_json_task():
    return {"status": "success", "data": "该任务返回值会走json序列化"}

# 未传参的任务默认沿用全局pickle配置
@app.task
def default_task():
    return {"status": "success", "data": "该任务返回值走全局pickle序列化"}

如果需要单独指定任务入参的序列化方式，同理在装饰器中传入task_serializer参数即可，比如@app.task(task_serializer='json')代表该任务接收的入参用json序列化。

2. 类形式定义的任务 #

直接在任务类中把序列化器设为类属性即可：

@app.task
class CustomSerializerTask(app.Task):
    result_serializer = 'json' # 单独指定结果序列化器
    task_serializer = 'json' # 单独指定入参序列化器

    def run(self, *args, **kwargs):
        return {"msg": "类定义任务同样支持单任务级序列化配置"}

注意事项 #

• 所有单任务用到的非默认序列化器，都必须提前加入全局accept_content白名单，否则Celery会出于安全考虑拒绝处理对应序列化格式的内容
• 任务级配置优先级最高，会完全覆盖全局同名配置，不需要调整其他全局参数
• 支持传入Celery兼容的所有合法序列化器值，包括pickle、json、yaml、msgpack等，前提是对应依赖已经安装完成

内容的提问来源于stack exchange，提问作者dragsu