如何避免Avro JSON编码后的消息被二次编码？

方法一：用fastavro直接序列化整个包裹对象（推荐） #

既然消息队列要求的是符合CDCEvent Avro schema的消息，直接用fastavro处理整个wrap对象，完全避开json.dumps导致的二次编码问题。

步骤如下：

1. 定义CDCEvent的Avro schema：

cdc_schema = {
    "type": "record",
    "namespace": "CDCEvent",
    "name": "CDCEvent",
    "fields": [
        {"doc": "The system that generated the event","type": "string","name": "sys"},
        {"doc": "The operation performed on the event","type": "string","name": "op"},
        {"doc": "The content of the event","type": "string","name": "payload"}
    ]
}

2. 生成message_str后，用fastavro序列化wrap对象：

from io import StringIO
import fastavro
from datetime import datetime

# 生成符合业务schema的message_str（原有逻辑）
message = {
    "name": "any",
    "ingestion_ts": datetime.utcnow(),
    "values": {
        "amount": 5,
        "countries": ["se", "nl"],
        "source": {"name": "web", "url": "whatever"}
    }
}
avro_schema = ""  # 替换为你的实际业务Avro schema
fo = StringIO()
fastavro.json_writer(fo, avro_schema, [message])
message_str = fo.getvalue()

# 构建包裹对象并序列化
wrap = {
    "sys": "my_system",
    "op": "c",
    "payload": message_str
}

output = StringIO()
fastavro.json_writer(output, cdc_schema, [wrap])
wrap_str = output.getvalue()

这样生成的wrap_str严格符合队列要求的Avro schema，payload字段的字符串不会被二次转义，消费者可直接正确解析。

方法二：手动拼接JSON字符串（不推荐，易出错） #

如果必须用标准JSON序列化，可手动拼接字符串避免自动转义，但这种方式在字段包含特殊字符时容易破坏JSON结构：

import json

# 先生成message_str（原有逻辑）

sys_op_json = json.dumps({"sys": "my_system", "op": "c"})
# 拼接payload部分，保留原始JSON字符串格式
wrap_str = sys_op_json[:-1] + ', "payload": ' + json.dumps(message_str) + '}'

问题根源 #

原代码中message_str已经是JSON格式的字符串，json.dumps(wrap)会把它当作普通字符串处理，自动将内部双引号转义为\"。但队列期望payload是未被二次转义的JSON字符串，导致消费者解析时无法识别原始JSON结构。

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