如何从包含后续非JSON数据的流中读取完整JSON对象?
针对你的JSON流读取问题的解决方案
咱们先把两个问题拆解开来,逐个说清楚:
1. 从流中读取尽可能多的JSON数据
如果你的流里是连续的多个独立JSON对象(比如{"key1":1}{"key2":2}这种格式),常规JSON解析器会因为整体不是单个合法JSON结构报错。这时候可以用流式增量解析的思路:
很多语言的JSON库都提供了“部分解析”的API,比如Python里的json.JSONDecoder.raw_decode()——它会从给定内容里解析出第一个完整的JSON对象,同时返回解析后剩下的未处理内容。你可以循环调用这个方法,直到流里没有可解析的JSON为止。
举个Python的简单示例:
import json def read_all_json_from_stream(stream): decoder = json.JSONDecoder() buffer = b"" while True: chunk = stream.read(1024) # 按块读取,避免一次性加载大文件 if not chunk: break buffer += chunk while True: try: obj, idx = decoder.raw_decode(buffer.decode()) yield obj buffer = buffer[idx:] except json.JSONDecodeError: # 剩下的内容不够组成完整JSON,继续读下一块 break
这种方式能高效处理流式的多JSON场景,不用把整个流加载到内存里。
2. 读取后面跟随非JSON数据的单个JSON对象
这种场景的核心是精准匹配大括号的嵌套层级,因为常规解析器遇到后面的非JSON内容会直接报错,而且还要注意区分字符串里的大括号(比如你示例中"blahblah}{{"里的符号不算结构大括号)。
具体实现思路是手动追踪状态:
- 先找到第一个
{作为JSON的起始点 - 初始化层级计数器
depth=1,同时标记是否处于字符串模式in_string=False、是否处于转义状态escaped=False - 逐个字符读取,根据当前状态处理:
- 遇到
"且未转义:切换字符串模式 - 遇到
\且在字符串中:标记下一个字符是转义的 - 不在字符串模式时,遇到
{就加层级,遇到}就减层级,当层级回到0时,就是完整JSON的结尾
- 遇到
- 提取这段内容用标准JSON解析器解析,剩下的就是非JSON部分
针对你给出的示例:
{ "SomeData": "blahblah", "SubObject": { "SomeData": "blahblah}{{" } } ... and some more text (not JSON)
字符串里的}{{不会被计数,因为处于字符串模式,当层级计数器回到0时,正好定位到第二个},也就是JSON的结尾。
用Python写个实用的示例函数:
import json def extract_json_before_non_json(stream): buffer = [] depth = 0 in_string = False escaped = False start_found = False while True: char = stream.read(1).decode() if not char: # 流结束,没找到闭合大括号,返回已读内容 break if not start_found: if char == '{': start_found = True depth = 1 buffer.append(char) continue buffer.append(char) if escaped: escaped = False continue if char == '"': in_string = not in_string elif char == '\\' and in_string: escaped = True elif not in_string: if char == '{': depth += 1 elif char == '}': depth -= 1 if depth == 0: # 找到完整JSON,停止读取 break json_str = ''.join(buffer) try: parsed_json = json.loads(json_str) remaining_content = stream.read().decode() return parsed_json, remaining_content except json.JSONDecodeError: # 解析失败,返回原始内容和剩余部分 return None, json_str + stream.read().decode()
这个函数能准确识别嵌套的JSON结构,不会被字符串里的大括号干扰。
内容的提问来源于stack exchange,提问作者user626528
相关产品推荐
相关产品推荐

