基于Python raw_decode实现连续拼接JSON流的无损解析

本文由Gemini辅助撰写，示例代码已在Linux Python 3环境执行验证。

在网络套接字通信、管道数据同步或流式日志投递等系统交互中，发送端有时会将多个独立的 JSON 数据报文连续输出到同一个数据流中。如果这些连续记录之间缺乏类似换行符（如 NDJSON/JSON Lines 规范）这样明确的单字符分隔符，或者属于紧凑拼接、内嵌不定长度空白字符的数据序列，直接使用常规的 json.loads 将无法处理这种复合文本，并会抛出 json.JSONDecodeError: Extra data 异常。

针对无显式定长分帧或缺乏标准换行符的连续 JSON 数据流，Python 标准库 json 模块提供了底层类 json.JSONDecoder 及其核心方法 raw_decode。该方法能够从指定字符串的偏移量开始，精确识别并解析出第一个合法的 JSON 对象，并返回解析结果与下一个待读取字符的索引位置，从而支持构建健壮的流式游标解析器。

核心机制与解析逻辑

json.loads(s) 的实现本质上是在内部调用 decoder.decode(s)，该过程在解析出顶层数据结构后，会强制检查后续是否存在非空白字符；若存在未消费字符则直接报错。而 JSONDecoder.raw_decode(s, idx=0) 允许显式传递起始解析游标 idx。

调用 raw_decode 时，解码器会按 JSON 语法规则从 idx 开始扫描，直至完整闭合当前顶层数据结构（如匹配的花括号或方括号、字面量值）。方法执行成功后返回一个二元元组 (obj, end_idx)：

1. obj：成功反序列化生成的 Python 对象（如字典、列表等）；
2. end_idx：当前 JSON 数据结构结束位置的下一个字符索引，可直接作为下一轮解析的参考基准。

需要注意的是，raw_decode 不会自动跳过不同 JSON 对象之间可能存在的空白填充字符（如空格、制表符、换行符）。因此，上层调度逻辑必须主动维护偏移游标，在进入下一次 raw_decode 调用前跳过间隔空白。

完整可执行代码示例

以下示例展示了如何基于游标轮询与空白清理，对多段连续拼接的 JSON 报文进行无损提取，并演示了在遇到尾部残缺报文时的错误定位行为：

import json

raw_payload_stream = (
    '{"action": "sync", "seq": 1}'
    '  {"action": "heartbeat", "seq": 2} '
    '{"action": "ack", "seq": 3, "payload": {"status": "success"}}'
)

decoder = json.JSONDecoder()
parsed_records = []
cursor = 0
total_length = len(raw_payload_stream)

while cursor < total_length:
    # 跳过记录间的空白与换行分隔符
    while cursor < total_length and raw_payload_stream[cursor].isspace():
        cursor += 1
    if cursor >= total_length:
        break

    # 从当前指针位置无损解析单个完整 JSON 结构
    record, next_idx = decoder.raw_decode(raw_payload_stream, idx=cursor)
    parsed_records.append(record)
    cursor = next_idx

# 验证解析批次数量与内部键值结构
assert len(parsed_records) == 3
assert parsed_records[0] == {"action": "sync", "seq": 1}
assert parsed_records[1] == {"action": "heartbeat", "seq": 2}
assert parsed_records[2]["payload"]["status"] == "success"
assert cursor == total_length

# 验证残缺非合法 JSON 串在解析时的异常定位行为
incomplete_fragment = '{"valid": 100} broken_trailing'
test_records = []
test_cursor = 0
obj, test_cursor = decoder.raw_decode(incomplete_fragment, idx=test_cursor)
test_records.append(obj)

assert len(test_records) == 1
assert test_records[0] == {"valid": 100}

try:
    # 跳过空白后尝试解析未识别标记应抛出 JSONDecodeError
    while test_cursor < len(incomplete_fragment) and incomplete_fragment[test_cursor].isspace():
        test_cursor += 1
    decoder.raw_decode(incomplete_fragment, idx=test_cursor)
    raise AssertionError("未捕获语法错误")
except json.JSONDecodeError as err:
    assert err.pos == test_cursor

print("MATERIAL_OK")

代码运行与输出验证

在 Linux Python 3 隔离环境中执行上述脚本，控制台输出如下：

MATERIAL_OK

所有断言均顺利通过，确认了以下执行特征：

1. 连续的 3 个独立字典对象被依次解析，即使对象之间穿插了不规则空格，游标仍精准向前推进至字符串尾部；
2. 嵌套结构（如 payload.status）被完整保留；
3. 面对后半段非法语法 broken_trailing 时，raw_decode 抛出 json.JSONDecodeError，并且异常对象上的 err.pos 属性精确指向非法语法的首字符位置 test_cursor。

技术边界与异常排查

在将该模式应用于长周期运行的数据处理管道时，需要注意以下边界与排查要点：

1. 前导与间隔空白的处理：raw_decode 不负责忽略对象外部的游离空白。如果游标推进后指向空格且未做跳过处理，直接调用 raw_decode 会因起始字符不合法而触发异常。因此必须在每次解析前通过 isspace() 驱动游标。
2. TCP 分包与半包缓冲截断：在网络套接字流式接收时，报文可能在 JSON 结构内部发生截断（例如仅接收到 {"action": "ack", "seq":）。调用 raw_decode 会引发 JSONDecodeError: Unterminated string 或 Expecting value。调用方必须捕获该异常，并将未消费的截断切片保留在接收缓冲区中，等待后续数据到达拼接后再继续尝试解码。
3. 大字符串切片与内存拷贝：传递给 raw_decode 的参数应尽量复用字符串引用并通过 idx 传参，避免在应用层频繁使用 s = s[next_idx:] 进行子串切片操作，以减少不必要的高频字符串拷贝开销。

官方文档参考：

• Python 标准库 json.JSONDecoder.raw_decode: Python 3 官方文档（对应模块与章节）