如何让Celery+Kombu改用simplejson替代标准库json模块？

环境信息 #

tact@tact_pub_api:/app$ python3.10 -m pip list -v | egrep -i 'celery|kombu'
celery                  5.2.3       /usr/local/lib/python3.10/site-packages pip
kombu                   5.2.4       /usr/local/lib/python3.10/site-packages pip
tact@tact_pub_api:/app$ python2.7 -m pip list -v | egrep -i 'celery|kombu'
DEPRECATION: Python 2.7 reached the end of its life on January 1st, 2020. Please upgrade your Python as Python 2.7 is no longer maintained. pip 21.0 will drop support for Python 2.7 in January 2021. More details about Python 2 support in pip can be found at https://pip.pypa.io/en/latest/development/release-process/#python-2-support pip 21.0 will remove support for this functionality.
celery                        3.1.26.post2 /usr/local/lib/python2.7/dist-packages pip
kombu                         3.0.37       /usr/local/lib/python2.7/dist-packages pip

问题与原因 #

从Python2.7迁移到3.10过程中，出现序列化报错：

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/kombu/serialization.py", line 39, in _reraise_errors
    yield
  File "/usr/local/lib/python3.10/site-packages/kombu/serialization.py", line 210, in dumps
    payload = encoder(data)
  File "/usr/local/lib/python3.10/site-packages/kombu/utils/json.py", line 68, in dumps
    return _dumps(s, cls=cls or _default_encoder,
  File "/usr/local/lib/python3.10/json/__init__.py", line 238, in dumps
    **kw).encode(obj)
  File "/usr/local/lib/python3.10/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.10/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.10/site-packages/kombu/utils/json.py", line 58, in default
    return super().default(o)
  File "/usr/local/lib/python3.10/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type bytes is not JSON serializable

原因是新版Kombu默认使用Python标准库json模块，而标准库json无法直接序列化bytes类型；旧版Celery+Kombu默认使用PyPI的simplejson，该模块支持bytes的序列化/反序列化，迁移期间需要保持兼容性。

验证示例：

>>> import json
>>> s = b'abc'
>>> json.dumps(s)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.10/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/local/lib/python3.10/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.10/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.10/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type bytes is not JSON serializable

解决方案 #

1. 安装simplejson #

先确保安装simplejson包：

pip install simplejson

2. 配置Celery使用simplejson #

有两种可靠的配置方式：

方式一：替换Kombu的json工具函数

在Celery应用初始化代码中添加：

import simplejson
from kombu.utils import json

# 替换默认的loads/dumps为simplejson的实现
json.loads = simplejson.loads
json.dumps = simplejson.dumps

方式二：重新注册json序列化器

通过Kombu的序列化注册机制，覆盖默认的json序列化器：

import simplejson
from kombu.serialization import register
from celery import Celery

app = Celery('your_app_name')

# 自定义simplejson的序列化/反序列化逻辑
def json_dumps(obj, **kwargs):
    return simplejson.dumps(obj, **kwargs)

def json_loads(s, **kwargs):
    return simplejson.loads(s, **kwargs)

# 注册自定义的json序列化器
register(
    'json', 
    json_dumps, 
    json_loads, 
    content_type='application/json', 
    content_encoding='utf-8'
)

# 配置Celery使用json序列化器
app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json'
)

3. 验证 #

配置完成后，测试bytes类型的序列化：

>>> import simplejson
>>> s = b'abc'
>>> simplejson.dumps(s)
'"abc"'
>>> simplejson.loads('"abc"')
'abc'

此时Celery任务传递bytes类型数据将不再报错，与旧版Python2.7环境的兼容性得以保障。

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