本教程旨在解决接收到多个json对象以非标准格式（即没有外部数组括号和逗号分隔）直接连接的场景。我们将介绍一种python解析策略，通过识别json对象的结束和开始标记来精确分割数据流，从而实现对每个独立json对象的成功解析和处理。
理解非标准json数据流
在处理API响应或日志数据时，我们通常期望接收到符合RFC 8259标准的JSON格式，例如单个JSON对象或一个包含多个JSON对象的数组。然而，在某些情况下，可能会遇到一种非标准的数据流，其中多个独立的JSON对象直接连接在一起，没有外部的方括号 [] 包裹，也没有逗号 , 进行分隔。这种格式使得标准的 json.loads() 函数无法直接解析整个字符串，因为它不是一个有效的JSON文档。
以下是一个典型的非标准JSON数据流示例：
{ "self": "https://example1.com", "key": "keyOne", "name": "nameOne", "emailAddress": "mailOne", "avatarUrls": { /* ... */ }, "displayName": "displayNameOne", "active": true, "timeZone": "Europe", "locale": "en_UK" } { "self": "https://example2.com", "key": "keyTwo", "name": "nameTwo", "emailAddress": "mailTwo", "avatarUrls": { /* ... */ }, "displayName": "displayNameTwo", "active": false, "timeZone": "Europe", "locale": "en_US" }在这种结构中，一个JSON对象的结束符 } 之后紧跟着下一个JSON对象的开始符 {，它们之间可能只有换行符。我们的目标是识别这些边界，并将数据流分割成独立的、可解析的JSON字符串。
解析策略：基于行识别边界
由于每个JSON对象都是一个独立的实体，其结束标志是 }，而下一个对象的开始标志是 {。我们可以利用这一特性，逐行扫描整个数据流。当检测到当前行是 { 并且前一行是 } 时，就意味着我们找到了两个JSON对象之间的边界。
立即学习“Python免费学习笔记（深入）”；
具体的解析步骤如下：
- 将整个非标准JSON数据流按行分割。
- 遍历这些行，维护一个当前正在构建的JSON对象的起始索引。
- 当发现一个行以 } 结束，并且紧接着的下一行以 { 开始时，这表明前一个完整的JSON对象已经结束。此时，将从起始索引到当前行（包含）的所有行拼接起来，形成一个完整的JSON字符串。
- 使用 json.loads() 函数解析这个字符串，并将其添加到结果列表中。
- 更新起始索引为当前行的下一行，继续处理剩余数据。
- 处理完所有行后，最后一个JSON对象需要单独处理。
python实现示例
下面是使用Python实现上述解析策略的示例代码：
import json # 示例非标准JSON数据流 data_stream = ''' { "self": "https://example1.com", "key": "keyOne", "name": "nameOne", "emailAddress": "mailOne", "avatarUrls": { "48x48": "https://test.com/secure/useravatar?avatarId=1", "24x24": "https://test.com/secure/useravatar?size=small&avatarId=1", "16x16": "https://test.com/secure/useravatar?size=xsmall&avatarId=1", "32x32": "https://test.com/secure/useravatar?size=medium&avatarId=1" }, "displayName": "displayNameOne", "active": true, "timeZone": "Europe", "locale": "en_UK" } { "self": "https://example2.com", "key": "keyTwo", "name": "nameTwo", "emailAddress": "mailTwo", "avatarUrls": { "48x48": "https://test.com/secure/useravatar?avatarId=2", "24x24": "https://test.com/secure/useravatar?size=small&avatarId=2", "16x16": "https://test.com/secure/useravatar?size=xsmall&avatarId=2", "32x32": "https://test.com/secure/useravatar?size=medium&avatarId=2" }, "displayName": "displayNameTwo", "active": false, "timeZone": "Europe", "locale": "en_US" } ''' def parse_concatenated_json(data_string: str) -> list: """ 解析包含多个直接连接的非标准JSON对象的字符串。 Args: data_string: 包含非标准JSON对象流的字符串。 Returns: 一个包含所有解析出的JSON字典的列表。 """ json_objects = [] lines = data_string.strip().splitlines() # 移除首尾空白并按行分割 current_object_start_line = 0 # 记录当前JSON对象开始的行索引 for i, line in enumerate(lines): # 检查当前行是否为"{"且前一行是否为"}" # 并且确保不是第一个JSON对象的开始（i > 0） if i > 0 and line.strip() == "{" and lines[i-1].strip() == "}": # 找到一个JSON对象的边界，解析前一个对象 json_segment = "n".join(lines[current_object_start_line:i]) try: json_objects.append(json.loads(json_segment)) except json.JSONDecodeError as e: print(f"解析JSON片段失败: {e}n片段内容:n{json_segment}") # 根据实际需求选择是跳过、记录错误还是中断 pass # 更新下一个JSON对象的起始行 current_object_start_line = i # 处理最后一个JSON对象 if current_object_start_line < len(lines): json_segment = "n".join(lines[current_object_start_line:]) try: json_objects.append(json.loads(json_segment)) except json.JSONDecodeError as e: print(f"解析最后一个JSON片段失败: {e}n片段内容:n{json_segment}") return json_objects # 调用解析函数 parsed_data = parse_concatenated_json(data_stream) # 打印解析结果以验证 print(f"成功解析 {len(parsed_data)} 个JSON对象:") for obj in parsed_data: print(json.dumps(obj, indent=2, ensure_ascii=False)) # 格式化输出，方便阅读代码解析：
- data_stream.strip().splitlines(): 首先使用 strip() 移除整个字符串开头和结尾的空白字符（包括换行），然后使用 splitlines() 将字符串分割成一个行的列表。
- current_object_start_line: 这个变量用于标记当前正在构建的JSON对象的起始行索引。
- for i, line in enumerate(lines): 遍历每一行及其索引。
- if i > 0 and line.strip() == “{” and lines[i-1].strip() == “}”: 这是核心判断逻辑。它检查：
- i > 0: 确保不是数据流的起始，因为第一个JSON对象之前没有 }。
- line.strip() == “{“: 当前行去除空白后是否为 {。
- lines[i-1].strip() == “}”: 前一行去除空白后是否为 }。
- 如果这三个条件都满足，则表明我们已经跨越了一个完整的JSON对象的边界。
- json_segment = “n”.join(lines[current_object_start_line:i]): 将从上一个起始行到当前行之前的所有行拼接起来，形成一个完整的JSON字符串。使用 “n”.join() 保持原始的换行符，这对于 json.loads 解析多行JSON字符串是必要的。
- json.loads(json_segment): 使用Python内置的 json 模块解析这个片段。
- current_object_start_line = i: 更新起始行索引为当前行的索引，为解析下一个JSON对象做准备。
- 处理最后一个JSON对象: 循环结束后，最后一个JSON对象仍然在 lines 列表中，从 current_object_start_line 到列表末尾。因此，需要一个额外的步骤来解析它。
注意事项与总结
- 数据格式依赖性: 这种方法高度依赖于每个JSON对象以 } 结尾，并且下一个对象以 { 开始的特定模式。如果数据流中存在其他分隔符或更复杂的结构（例如，JSON对象内部的字符串也包含 { 或 } 但不在行首或行尾），则此方法可能需要调整或失效。
- 错误处理: 在实际应用中，应加入更健壮的错误处理机制。例如，如果 json.loads() 失败，应该捕获 json.JSONDecodeError 异常，并决定是跳过该片段、记录错误信息，还是终止解析过程。示例代码中已加入了基本的错误捕获。
- 性能考量: 对于非常大的数据流，逐行读取和字符串拼接可能会消耗较多的内存和CPU。如果性能是关键因素，可以考虑使用更流式（streaming）的解析器，或者在拼接字符串时优化内存使用。
- 数据清洗: 在 splitlines() 之前使用 strip() 是一个好习惯，可以去除整个数据流开头和结尾可能存在的无关空白字符。同样，在比较行内容时使用 line.strip() 可以忽略行内的前导/尾随空白。
通过上述基于行识别边界的策略，我们可以有效地解析和处理那些不符合标准json数组格式，而是以非标准方式直接连接的多个JSON对象流。这种方法为处理来自不规范数据源的JSON数据提供了一个实用的解决方案。