4. JSON数据解析:json.loads与json.dumps实战、嵌套JSON处理、异常数据捕获

遥测数据在网络上传输,最常用的格式就是 JSON。说白了,它就是一种轻量级的数据交换格式,人类能读懂,机器也好解析。但实际生产环境里,你收到的 JSON 可不总是规规矩矩的。格式错乱、字段缺失、类型不匹配……这些问题我几乎每天都会遇到。

这一章,我们就来啃下 JSON 解析这块硬骨头。我会结合自己在遥测系统里的实战经验,把 json.loads、json.dumps、嵌套处理以及异常捕获这些知识点,一个一个讲透。

4.1 核心函数:json.loads 与 json.dumps

Python 标准库的 json 模块,提供了两个最核心的函数。一个负责把 JSON 字符串转成 Python 对象,另一个负责把 Python 对象转成 JSON 字符串。理解它们,是处理遥测数据的第一步。

4.1.1 json.loads:从字符串到Python对象

json.loads 接收一个 JSON 格式的字符串,返回对应的 Python 数据类型。映射关系很简单:

JSON 类型 Python 类型
object dict
array list
string str
number (int) int
number (real) float
true / false True / False
null None

看个最简单的例子:

import json

# 模拟从遥测数据流中收到的一条消息
raw_message = '{"device_id": "sensor-01", "temperature": 25.6, "status": "online"}'

# 解析成 Python 字典
parsed = json.loads(raw_message)
print(parsed["device_id"])   # 输出: sensor-01
print(parsed["temperature"]) # 输出: 25.6

嗯,这里要注意。如果字符串本身不是合法的 JSON,json.loads 会直接抛出 json.JSONDecodeError。我在项目里见过不少新手,解析前不做校验,结果程序一跑就崩。

4.1.2 json.dumps:从Python对象到字符串

反过来,json.dumps 把 Python 对象序列化成 JSON 字符串。这在发送遥测数据、或者把处理结果写回文件时特别常用。

import json

data = {
    "device_id": "sensor-01",
    "temperature": 25.6,
    "status": "online"
}

# 序列化成 JSON 字符串
json_str = json.dumps(data)
print(json_str)
# 输出: {"device_id": "sensor-01", "temperature": 25.6, "status": "online"}

我个人习惯在调试时加上 indent=2 参数,让输出更易读:

print(json.dumps(data, indent=2))

输出效果:

{
  "device_id": "sensor-01",
  "temperature": 25.6,
  "status": "online"
}
小技巧: 如果 Python 对象里包含 datetime 等非 JSON 原生类型,直接 dumps 会报错。这时候可以自定义 default 参数,或者用 str() 先转成字符串。

4.2 嵌套JSON处理:从扁平到深层

遥测数据很少是扁平的。你想想看,一个设备上报的数据,可能包含多个传感器、每个传感器又有自己的配置和读数。这就形成了嵌套结构。

比如下面这个例子:

{
  "gateway_id": "gw-001",
  "timestamp": "2025-03-21T10:30:00Z",
  "sensors": [
    {
      "type": "temperature",
      "value": 25.6,
      "unit": "celsius"
    },
    {
      "type": "humidity",
      "value": 68.2,
      "unit": "percent"
    }
  ]
}

解析这种嵌套 JSON,核心思路就是一层一层往下取。用 Python 的字典和列表操作就能搞定:

import json

raw = '''
{
  "gateway_id": "gw-001",
  "timestamp": "2025-03-21T10:30:00Z",
  "sensors": [
    {"type": "temperature", "value": 25.6, "unit": "celsius"},
    {"type": "humidity", "value": 68.2, "unit": "percent"}
  ]
}
'''

data = json.loads(raw)

# 获取网关ID
gw_id = data["gateway_id"]

# 遍历传感器列表
for sensor in data["sensors"]:
    s_type = sensor["type"]
    s_value = sensor["value"]
    print(f"{s_type}: {s_value}")

输出:

temperature: 25.6
humidity: 68.2

我在项目中遇到过更深的嵌套,比如五层甚至六层。那时候如果还用 data["a"]["b"]["c"]["d"] 这样硬编码,代码会变得又臭又长。我的建议是:

  • 用 .get() 方法代替直接索引,避免 KeyError。
  • 提取公共路径,比如写一个辅助函数,专门处理深层取值。
  • 考虑用 jsonpath-ng 库,类似 XPath 那样定位节点,特别适合复杂嵌套。
核心原则: 嵌套 JSON 解析,宁可多写几行校验,也不要赌数据一定完整。生产环境里,字段缺失是常态。

4.3 异常数据捕获:让程序更健壮

遥测数据从设备端经过网络传输,中间可能被截断、篡改、或者干脆就是格式错误。如果不做异常捕获,一个坏数据就能让整个解析流程瘫痪。

我曾经遇到过一件事:某天凌晨,告警系统突然疯狂报警,说所有设备都离线了。排查了半天,发现是一条 JSON 数据里多了一个逗号,导致解析失败,整个处理线程卡死了。从那以后,我在所有 JSON 解析的地方都加了异常处理。

4.3.1 捕获 JSONDecodeError

这是最常见的异常。当字符串不是合法的 JSON 时,就会抛出它。

import json

def safe_parse(raw_string):
    try:
        return json.loads(raw_string)
    except json.JSONDecodeError as e:
        print(f"JSON 解析失败: {e}")
        # 记录日志,或者返回默认值
        return None

# 测试坏数据
bad_data = '{"device_id": "sensor-01", "temperature": 25.6, }'  # 多了一个逗号
result = safe_parse(bad_data)
print(result)  # 输出: None

4.3.2 捕获 KeyError 和 TypeError

即使 JSON 解析成功了,也不代表数据是完整的。比如字段缺失,或者类型不对。

def extract_temperature(data):
    try:
        return data["temperature"]
    except KeyError:
        print("字段 'temperature' 缺失")
        return None
    except TypeError:
        print("data 不是字典类型")
        return None

4.3.3 统一异常处理模式

在实际的遥测过滤系统中,我通常会写一个统一的解析函数,把各种异常都兜住:

def robust_json_parse(raw_string):
    """
    健壮的 JSON 解析函数
    返回: (解析成功标志, 解析后的数据或错误信息)
    """
    if not isinstance(raw_string, str):
        return False, "输入不是字符串类型"
    
    try:
        data = json.loads(raw_string)
    except json.JSONDecodeError as e:
        return False, f"JSON 格式错误: {e}"
    
    if not isinstance(data, dict):
        return False, "JSON 根节点不是对象"
    
    return True, data

这样调用方就能根据返回的布尔值,决定是继续处理还是跳过这条数据。

注意: 不要在所有地方都用 except Exception 一把抓。那样会掩盖真正的 bug。尽量精确捕获你预期会发生的异常类型。

4.4 知识体系总览

下面这张图,把本章的核心知识点串了起来。你可以把它当作一个快速参考。

JSON数据解析知识体系 JSON 数据解析知识体系 JSON 数据解析 json.loads json.dumps 嵌套 JSON 处理 字符串 → Python 对象 类型映射表 Python 对象 → 字符串 indent 参数美化 逐层索引 / .get() 辅助函数提取 异常数据捕获(JSONDecodeError / KeyError / TypeError)

4.5 实战建议

最后,分享几个我在实际项目中沉淀下来的经验:

  • 永远不要信任外部输入。 不管是来自设备、文件还是网络,JSON 数据在解析前都可能是坏的。
  • 日志要打全。 捕获异常后,把原始数据和异常信息都记录下来。不然出了问题,你连复现都难。
  • 考虑性能。 如果每秒要解析成千上万条 JSON,json.loads 的性能开销就不能忽略。这时候可以考虑用 orjson 或 ujson 这类第三方库,速度能快好几倍。
  • 写单元测试。 准备几组典型的坏数据,确保你的解析函数能正确处理。我每次改解析逻辑,都会先跑一遍测试用例。

JSON 解析看起来简单,但坑都在细节里。把这一章的内容吃透,你在遥测数据处理的路上就能少踩很多坑。