搞开发这些年,我几乎每天都在跟JSON打交道。项目从初始框架搭建到上线维护,配置文件是JSON,接口返回是JSON,日志结构化也是JSON……可以这么说,JSON文件读写已经不光是初学者绕不开的基本功,更是日常开发中最容易被“想当然”的高频操作。很多人觉得无非就是json.load和json.dump两个函数,但真到了生产环境,编码问题、异常处理、大文件性能、嵌套结构校验,每一个环节都能让人翻车。
这篇文章我打算从一个实际开发者踩坑的角度,把JSON文件读写这件事完整拆开讲一遍。不绕弯子,直接用Python作为主语言演示,里面穿插我在项目里积累的操作习惯和排查经验。无论你刚接触JSON,还是写了好几年代码偶尔还会被JSONDecodeError折磨,这篇都能给你一些值得直接抄走的处理方案。
1. JSON文件读写的基本功,熟练到什么程度才算过关
1.1 先搞清楚JSON和Python字典到底差在哪
很多人会下意识觉得JSON就是Python的字典,其实这个印象害了不少人。JSON全称是JavaScript对象表示法,它本质上是一种文本格式,用来在不同语言、不同系统之间传递结构化数据。而Python字典是内存中的对象,它有自己的数据类型和方法。两者之间的关系靠序列化和反序列化来桥接:把Python对象变成JSON文本叫序列化,把JSON文本还原成Python对象叫反序列化。
这个区别在文件读写上体现得非常直接。你打开一个.json文件,看到里面那些花括号、方括号、冒号,全都是字符串符号;只有当你用json.load()把文件内容读进内存之后,它才变成Python的dict、list、int这些真实对象。反过来说,你往文件里json.dump()数据时,Python对象也被转换成了纯文本。
我见过不少新手直接用open()读JSON文件,然后拿eval()去解析。这种做法不仅危险,还会因为JSON和Python语法细节的差异(比如true、false、null和Python的True、False、None不一样)产生一堆莫名其妙的问题。标准做法永远是通过json模块来处理,这件事没有任何商量余地。
1.2 JSON支持的数据类型和Python并不完全对等
JSON里只有六种基础类型:对象、数组、字符串、数字、布尔值和空值。而Python的数据类型远比这个丰富,比如元组、集合、日期时间对象,JSON都不认识。这就造成了一张并不完全一一对应的映射表,我整理了一下平时最常用的对应关系:
| JSON类型 | Python类型 | 注意事项 |
|---|---|---|
对象{} | 字典dict | 键必须是字符串,读取后会自动变成Python字符串 |
数组[] | 列表list | 不会映射成元组,即使你写入时用的是元组 |
字符串"..." | 字符串str | JSON强制双引号,单引号会导致解析失败 |
| 数字 | int/float | JSON不区分整数和小数,解析由Python自动判断 |
布尔值true/false | bool | 小写开头,和Python的True/False写法不同 |
空值null | None | 读取后变成None,写入时None也会变成null |
这张表看起来简单,实际坑不少。最典型的就是元组写入后会变成列表。比如你有一个坐标点(3, 5),直接json.dump()存进文件,再读出来就变成[3, 5]。如果业务代码里强依赖元组特性,就会在后续判断时报错。所以设计数据结构时,能不用元组序列化就不要用,或者读取后手动做一次转换。
另一个值得关注的是字典的键。JSON规范里键必须是字符串,Python的json模块写入时会自动把键转成字符串。假如你用整数1做键,写入的是"1",读出来也是字符串"1",如果代码里还用data[1]去取,就会得到KeyError。这个细节在从外部接口接收JSON时尤其容易踩中,因为别的系统可能生成了纯字符串键的字典。
1.3 一个合格的JSON读写操作必须考虑文件编码
JSON文件本质是文本文件,所以读写时一定要关注编码。现在主流标准是UTF-8,但如果对接老系统,也可能遇到GBK、GB2312,甚至带BOM标记的UTF-8文件。很多人写代码时从不指定encoding参数,完全依赖操作系统的默认编码。在Linux上默认是UTF-8通常没问题,到了Windows上默认可能是GBK,同样的代码换个环境就乱码。
我的习惯是只要打开文本文件,就显式写上encoding="utf-8"。虽然多打几个字符,却省掉了一类非常隐蔽的跨平台问题。如果你要读取的文件来自Windows下的老工具,还需要处理BOM头,utf-8-sig这个编码别名就是专门干这个的,它会自动去掉文件开头可能存在的\ufeff字符。这些细节后面章节会展开讲。
2. 读取JSON文件:不同需求下的正确打开方式
2.1 最直接的json.load全量读入
先说最常规的场景:项目里有一个配置文件,或者一个不算太大的数据文件,你希望在程序启动时一次性把所有数据读进内存。这时直接用json.load()即可。
import json with open("config.json", "r", encoding="utf-8") as f: config = json.load(f) print(config["app_name"])这段代码的逻辑是:先用open()拿到一个文件对象,再把它交给json.load(),由json模块负责读取文件内容并完成解析。整个流程干净利落,适合绝大多数中小文件。
值得说明的是,json.load()内部会先读取全部文件内容到内存,再执行解析,所以它本质上是一个全量加载方案。如果文件有几十MB甚至上百MB,这种读法会带来明显的内存压力。我参与过的某日志分析项目,单个JSON文件有200多MB,用json.load()直接读入,内存占用一度冲到1GB以上,就是因为解析过程中不仅保存了原始文件字符串,还创建了一整套完整的Python嵌套对象。
全量读入的优势是使用简单、访问灵活。读完之后,你可以像操作普通字典一样对数据做任意修改、检索和统计。对配置文件、元数据、中小型业务数据来说,这是最推荐的方案。
2.2 从字符串解析:json.loads的适用场景
json.loads()和json.load()只差一个字母,却对应完全不同的入口。loads()接收的是字符串,不是文件对象。这个函数最常用的场景是从网络请求、消息队列或数据库字段里拿到JSON文本,需要转成Python对象。
import json raw_json = '{"name": "某项目", "version": "1.2.0", "enabled": true}' data = json.loads(raw_json) print(data["name"]) # 某项目 print(data["enabled"]) # True注意字符串里的true不是Python的True,因为json.loads()在解析时会把JSON规范的布尔值正确映射成Python布尔值。有些开发者图省事,想用ast.literal_eval()来解析JSON字符串,虽然某些简单场景能跑通,但一旦遇到null、true、false这些JSON特有写法,结果往往就不对了。老老实实用json.loads(),是兼容性最好的方案。
还有一类场景是文件来源不太可控,比如用户上传了一个JSON文件,你想先校验内容再决定是否处理。这时候可以先把整个文件读成字符串,再用json.loads()解析,在解析处加异常捕获。这样能更灵活地控制错误的处理逻辑,而不是让异常直接冒泡到顶层。
2.3 空文件、UTF-8编码和BOM标记的处理
读取JSON文件时最常见的三个环境问题,我放到一起说。
第一个是空文件。如果文件是0字节,json.load()会直接抛JSONDecodeError,提示Expecting value: line 1 column 1 (char 0)。这种情况经常出现在程序第一次运行时,程序想读取还不存在的配置文件。我的处理方案是先用os.path.exists()判断文件是否存在,再判断文件大小,或者干脆在异常处理分支里给出一份默认配置。
import os import json def load_config(path, default_config): if not os.path.exists(path) or os.path.getsize(path) == 0: return default_config with open(path, "r", encoding="utf-8") as f: try: return json.load(f) except json.JSONDecodeError: return default_config第二个是编码不匹配。文件实际是GBK编码,但你用UTF-8去读,就会抛UnicodeDecodeError。这种情况在对接老旧Windows工具生成的文件时经常遇到。如果项目确实要兼容多编码,可以先用二进制模式读取文件,再尝试用不同编码解码,或者使用chardet这类库做编码探测。不过我要提醒一句:在生产环境里,最好的策略不是做智能识别,而是统一约定UTF-8,在上游就把编码问题解决掉。
第三个是BOM标记。UTF-8本不需要BOM,但Windows上一些编辑器保存文件时会自动加上。BOM在open()读取时不会自动被移除,而是会被当成字符串的一部分。结果就是json.load()解析时在第一个字符处报错。要兼容这种情况,读文件时把编码指定为utf-8-sig即可,Python会替你处理掉BOM头。如果你的程序只允许标准UTF-8,那读取带BOM的文件就会失败,这也是一个合理的选项,取决于你对上游文件的控制力。
2.4 大JSON文件怎么读才不把内存撑爆
当JSON文件大到一定程度,全量读入就不现实了。这里分两种处理思路。
第一种是流式解析,适合JSON里是一个巨大的数组,每条记录相对独立的情况。比如一个用户行为日志文件,顶层是一个几百万条记录的大数组。用json.load()会把整个数组都加载进内存,但如果我只需要逐条处理数据,完全可以借助ijson这个第三方库做增量解析。
import ijson with open("huge.json", "rb") as f: parser = ijson.items(f, "item") for record in parser: process(record)这里的"item"是指顶层数组的每个元素。ijson会边读边解析,避免一次性构建完整的Python列表,内存占用能显著下降。我测过一个300MB的JSON文件,json.load()峰值内存大约在1.2GB,换成ijson逐条处理后峰值降到不到200MB,差距非常明显。
第二种是分块处理,如果JSON结构不适合流式解析,可以先用工具做预处理,比如把一个大JSON拆成多个小文件,或者转换成NDJSON格式。NDJSON也叫JSON Lines,每行一个独立JSON对象,处理起来比大数组方便得多,逐行读取即可,还能配合多进程做并行分析。
with open("data.ndjson", "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: record = json.loads(line) process(record)不过要强调一点:大文件优化应该发生在数据量真的变大之后。如果只是几十KB的配置,直接json.load()就是最优解,过早引入流式解析只会增加代码复杂度。
3. 写入JSON文件:从能写到写好
3.1 基础写入和最容易忽略的格式化问题
写入JSON文件,初学者通常能写出这样的代码:
import json data = {"name": "某项目", "version": "1.2.0", "features": ["auth", "report"]} with open("output.json", "w", encoding="utf-8") as f: json.dump(data, f)这段代码没有任何硬伤,但打开生成的文件会发现,所有内容被压缩成一行,没有缩进换行。如果文件是给程序读的,这没有问题;如果开发者需要人工查看或排错,这样的文件体验就很差。
为了让输出更可读,可以调整json.dump()的缩进参数。最常用的是indent=2和indent=4,我习惯用indent=2,在大部分编辑器和终端里显示效果都不错。
with open("output.json", "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False)indent参数会让输出的每一层结构增加换行和空格缩进。另外还要注意,任何修改配置文件的操作,最后都应该保证写入后的JSON格式合法。如果写入过程中程序崩溃,文件可能只写了一半,下次启动程序就会解析失败。这个问题后面的原子写入部分会专门说。
3.2 ensure_ascii=False为什么这么重要
很多人在第一次用json.dump()写中文时都会被吓到:明明数据里是"项目",生成的文件里却变成"\u9879\u76ee"。这不是bug,而是json模块的默认行为:为了兼容ASCII环境,它会默认把非ASCII字符转义成Unicode编码。
import json data = {"name": "项目"} with open("output.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=True) # 默认,输出 {"name": "\u9879\u76ee"} with open("output_utf8.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) # 输出 {"name": "项目"}如果你确定文件是UTF-8编码,并且希望人工可读,ensure_ascii=False是必须的。它不对中文做转义,直接保存原始字符。这个参数在接口返回JSON时更要谨慎,因为HTTP响应头通常已经指定了UTF-8,再转义成\uXXXX只会白白增大传输体积,还会让前端开发者头大。
值得一提的是,ensure_ascii对Python 3.7以上版本的行为没有变化,依然是默认True。这和其他语言默认输出中文原字符的习惯不太一样,所以从其他语言转过来的开发者也要注意。
3.3 写入模式:w和a的选择,以及文件覆盖风险
open()的写入模式选择是个容易想当然的地方。"w"模式会直接覆盖原文件,如果文件不存在就新建;"a"模式则是追加。JSON文件通常是整体读入、整体写出的结构,所以追加模式用得很少。但有一种场景例外:写日志时用JSON Lines格式,每条日志一行,这时候"a"模式就非常合适。
import json log_entry = {"level": "INFO", "message": "user login", "ts": "2025-05-01T10:00:00Z"} with open("app.ndjson", "a", encoding="utf-8") as f: f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")不要试图用"a"模式往一个标准JSON对象里追加内容,因为标准JSON要求一个文件里只有一个完整的顶层对象或数组,如果把两个JSON对象直接拼在一起,下次解析必然失败。如果确实要往同一个JSON文件里不断添加数据,最稳的做法还是读出来、改完、再整体写回去。
"w"模式还有一个隐患是覆盖时没有备份。我会在涉及配置文件的写入操作时自动生成一份.bak备份,一旦新配置格式错误或逻辑有问题,还能快速回滚。这个习惯在线上环境帮你挽回的局面可能远比你想象得多。
3.4 原子写入:防止程序崩溃导致JSON文件损坏
假设你要把一份重要配置写回文件,写了一半磁盘满了,或者进程被强杀,文件就会处于残缺状态。下一次程序启动读取配置时,大概率会直接崩溃。避免这种情况的经典方案是原子写入,也就是先写临时文件,再替换原文件。
import os import tempfile def atomic_write_json(path, data): dir_name = os.path.dirname(path) or "." fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp") try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) os.replace(tmp_path, path) except Exception: os.unlink(tmp_path) raise流程并不复杂:tempfile.mkstemp()会在目标文件同目录下生成一个临时文件,把数据完整写入临时文件,最后调用os.replace()把临时文件原子性地覆盖到目标位置。由于替换操作在同一文件系统内是原子的,要么旧文件完好,要么新文件完整,不存在中间状态。
我在实际项目中把这个逻辑封装成了公共函数,凡是写配置文件、缓存文件、状态文件,一律走这个通道。可能有人觉得多此一举,但分布式系统和定时任务里进程被杀是常态,养成这种习惯能少处理很多诡异问题。
4. JSON数据的结构设计与自定义对象序列化
4.1 嵌套结构读写时,如何优雅地取深层数据
真实项目里的JSON几乎不太可能是单层字典。大部分时候是字典套列表、列表套字典,一层套一层。读取这种嵌套结构,最怕的就是某个中间键不存在。
# 数据示例 { "service": { "name": "report-server", "backends": [ {"host": "10.0.0.1", "port": 8080}, {"host": "10.0.0.2", "port": 8081} ] } }如果我想取第二个后端的端口,用链式下标data["service"]["backends"][1]["port"]最直接,但一旦中间某个键缺失,就会抛出KeyError或IndexError。更稳的取法是用get()逐层降级。
backends = data.get("service", {}).get("backends", []) if len(backends) > 1: port = backends[1].get("port", 0) else: port = 0这种写法虽然啰嗦一些,但对不确定的数据结构非常友好。如果你要处理的JSON来自第三方接口,字段缺失、顺序变化都是常态,用get()加默认值能避免一堆异常分支。Python 3.8引入的海象表达式可以让多层get()略微紧凑一点,但可读性因人而异,看团队习惯即可。
另一个常用工具是jsonpath-ng这个库,相当于JSON世界的XPath,能通过表达式一次性定位到深层数据。
from jsonpath_ng import parse expr = parse("$.service.backends[*].port") ports = [match.value for match in expr.find(data)]当嵌套深度达到四五层时,用JsonPath表达式比逐层get()清晰得多。不过它毕竟引入了第三方依赖,对于简单的两层嵌套,未必值得。
4.2 自定义类的对象如何序列化成JSON
写业务代码时,我经常遇到这样的需求:程序里有一批自定义类的实例,希望保存成JSON存档,下次启动时恢复。直接json.dump()自定义对象会得到TypeError: Object of type Xxx is not JSON serializable。
解决这个问题最常见的做法是给json.dump()传入default参数,在回调函数里把对象转换成字典。
import json from datetime import datetime class Task: def __init__(self, task_id, title, created_at): self.task_id = task_id self.title = title self.created_at = created_at def task_to_dict(obj): if isinstance(obj, Task): return { "task_id": obj.task_id, "title": obj.title, "created_at": obj.created_at.isoformat(), } if isinstance(obj, datetime): return obj.isoformat() raise TypeError(f"Object of type {type(obj)} is not JSON serializable") task = Task(1001, "处理报表", datetime.now()) with open("task.json", "w", encoding="utf-8") as f: json.dump(task, f, default=task_to_dict, ensure_ascii=False, indent=2)反过来,读取时再根据业务需要把这个字典恢复成对象。如果对象结构简单,也可以直接使用内置的__dict__属性。
def task_to_dict(obj): return obj.__dict__用__dict__的好处是省事,缺点是会把所有实例属性都暴露出来,如果有不需要持久化的内部变量也会一起保存;另外如果类里包含不可序列化对象,比如文件句柄、锁对象,就会在序列化时报错。所以更推荐的做法是显式指定有哪些字段需要保存。这个决策应该根据对象结构的稳定程度来定。
4.3 一套可复用的配置读写模板
配置管理是最典型的JSON文件读写场景。我这里整理了一套经过多轮迭代后的配置读写模块。它的核心特点有三个:读取时自动填充默认值、写入时保留格式化、异常时给出明确错误信息。
import json import os class JsonConfig: def __init__(self, path, defaults=None): self.path = path self.defaults = defaults or {} self.data = {} self.load() def load(self): if not os.path.exists(self.path): self.data = json.loads(json.dumps(self.defaults)) return try: with open(self.path, "r", encoding="utf-8") as f: self.data = json.load(f) except json.JSONDecodeError as e: raise RuntimeError(f"配置文件语法错误: {self.path}, 错误位置: {e}") def save(self): tmp_path = self.path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(self.data, f, indent=2, ensure_ascii=False) os.replace(tmp_path, self.path) def get(self, key, default=None): return self.data.get(key, default) def set(self, key, value): self.data[key] = value self.save()这套模板在中小型服务里很实用。一个需要注意的点是load()里对默认配置的处理:json.loads(json.dumps(self.defaults))是为了深拷贝默认配置,避免默认字典被后续修改污染。很多人直接用self.defaults赋值,结果改配置时默认值也被改掉了,越改越乱。深拷贝这一步在配置系统里很关键,值得养成习惯。
5. 错误处理:JSON读写最常见的坑和排查路径
5.1 JSONDecodeError:文件解析失败的原因排查
JSONDecodeError是JSON文件读写中遇到最多的错误,没有之一。它不只是说“内容不是合法JSON”,而是帮你定位到了具体行列位置。排查这类错误时,我通常按下面几步走。
第一步,看错误信息给出的行列位置。JSONDecodeError的提示形如Expecting ',' delimiter: line 2 column 5 (char 42),它直接告诉你问题出在哪一行哪一列,先打开文件跳到对应位置看个究竟。
第二步,确认文件是否完整。分布式环境里,文件传输中断、磁盘配额不够,都会留下残缺文件。一个只有半个JSON对象的文件,最后一行往往少了一个}或]。遇到这种情况,可以先检查末尾字符,或者用json.tool命令行快速校验。
python -m json.tool config.json第三步,排查是否存在多余的逗号、单引号或注释。JSON规范非常严格:字符串必须双引号,对象最后一项后不能有逗号,不支持//注释。市面上很多所谓的JSON文件其实是超集,比如JSON5、带注释类型的配置文件,直接用json.load()必然报错。如果是前端项目流传出来的配置文件,很可能带注释,需要先做预处理。
第四步,确认是否混入了不可见字符。BOM头、控制字符、非UTF-8字符,都可能导致解析报错。读取时明确encoding="utf-8-sig"能解决一部分问题,如果依旧报错,可以用二进制模式读入,再检查前几个字节。
with open("unknown.json", "rb") as f: head = f.read(20) print(head.hex())看到ef bb bf开头就是BOM头,用utf-8-sig重新读取即可。
5.2 TypeError: Object of type X is not JSON serializable
这个错误通常出现在写入阶段,核心原因是JSON不支持Python特有的对象类型。常见的有:自定义类实例、datetime、date、Decimal、bytes、set。
解决思路是在序列化入口做类型转换。我习惯维护一个类型转换函数,统一处理项目里常用的类型,而不是每处写入前各写一遍转换逻辑。
import json from datetime import datetime, date from decimal import Decimal def default_serializer(obj): if isinstance(obj, (datetime, date)): return obj.isoformat() if isinstance(obj, Decimal): return float(obj) if isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") if isinstance(obj, set): return list(obj) raise TypeError(f"Type {type(obj)} not serializable")然后所有json.dumps()和json.dump()都传default=default_serializer。这个方案的优点是统一集中管理,新项目类型只需改一处。缺点是如果两个团队用了不同的转换函数,数据格式可能不一致,所以建议把它放到公共工具包里统一定义。
如果嵌套结构太深,排查哪个字段出了问题有个小技巧:先用default=str强制全部转成字符串,虽然数据可读性变差,但至少能定位是哪一层出了问题。这个方法在调试阶段偶尔会救急。
5.3 UnicodeDecodeError:编码不符和乱码处理
读取JSON文件时遇到UnicodeDecodeError,本质是读取使用的编码和文件实际编码不一致。最常见的两种情况:文件是GBK但用UTF-8读,或者文件是UTF-8但用了GBK读。
在无法确认编码的情况下,可以尝试用二进制模式读取,再用errors="replace"容错解析,先看看数据大概长什么样。
with open("data.json", "r", encoding="utf-8", errors="replace") as f: content = f.read()这样即使存在非法字节也不会抛异常,但你要意识到被替换的字符已经丢失了,后续解析不一定成功。更稳妥的办法是用chardet或者charset-normalizer去做编码检测。
import chardet with open("data.json", "rb") as f: raw = f.read() encoding = chardet.detect(raw)["encoding"] text = raw.decode(encoding) data = json.loads(text)编码检测不保证100%准确,尤其是短文件。所以我的原则仍然是在项目里统一UTF-8出口,对所有外部文件在进入系统前先做一次转码,不要在业务代码里到处做编码判断。
5.4 常见问题速查表
把高频问题汇总成一张表,排查时可以快速对照。
| 报错 | 大概率原因 | 处理思路 |
|---|---|---|
| JSONDecodeError: Expecting value | 空文件、文件只有空白字符 | 检查文件大小,用默认配置兜底 |
| JSONDecodeError: Expecting ',' delimiter | 结尾多逗号、对象未闭合 | 打开文件定位到行列 |
| JSONDecodeError: Expecting property name enclosed in double quotes | 单引号或键未加引号 | 确认文件是标准JSON,而非JS对象写法 |
| UnicodeDecodeError | 编码不匹配 | 用正确的编码读取,或做编码探测 |
| UnicodeDecodeError: 'utf-8' codec can't decode | BOM或非法字节 | 尝试utf-8-sig,检查文件头部字节 |
| TypeError: not JSON serializable | 对象里有自定义类型 | 提供default序列化函数 |
| RecursionError | JSON嵌套过深 | 检查是否存在循环引用,考虑限制深度 |
| json.dump写入后文件为空 | 写入异常或权限不足 | 检查异常捕获逻辑,必要时用原子写 |
还有一点我要特别提醒:json.dump()写入文件如果用到encoding参数是无效的,因为Python的json模块会以字符串形式接收数据,真正决定写盘编码的是open()时指定的encoding。也就是说json.dump(data, f, encoding="utf-8")这种写法虽然不报错,但encoding参数被忽略了,文件编码完全由open()决定。这个误解很常见,但很少有人注意到。
6. 性能优化:高频率与大文件下的JSON读写策略
6.1 高频写入时先合并IO而不是每次落盘
有些服务会把运行状态频繁写进JSON文件,比如每分钟记录一次指标。如果每次都打开文件、dump、关闭,文件系统I/O开销会非常大。以我观察到的经验,单线程下每秒重复执行JSON dump几十次,性能就开始明显下降,磁盘压力也上来了。
一种常见的优化思路是把写入频率降下来,改成定期批量落盘。例如服务持续收集指标,在内存里汇总成列表,每30秒才把所有数据写一次。这能显著减少I/O次数,代价是进程崩溃时会丢失最近一个批次的数据,需要根据业务对可靠性的要求来权衡。
import time import json buffer = [] last_flush_time = time.time() def collect_and_maybe_flush(item): global buffer, last_flush_time buffer.append(item) if time.time() - last_flush_time >= 30: with open("metrics.json", "w", encoding="utf-8") as f: json.dump(buffer, f, ensure_ascii=False) buffer = [] last_flush_time = time.time()如果连周期落盘都嫌慢,还可以把数据先追加到内存缓冲区,用另一个线程处理磁盘写入。但这样会带来线程安全、数据一致性等问题,复杂度直线上升。我的建议是:先确认性能瓶颈真的在JSON写入上,再去搞并发,不要为了优化而优化。
6.2 大文件流的增量解析比一次性load更稳
前面提到ijson可以做增量解析。在真实数据管道里,我还经常遇到每个JSON对象体积不大、但文件总条数极多的情况。这时候解析的单条数据很小,但顶层容器很大。使用ijson只取需要处理的字段,也能有效减少内存占用和解析开销。
import ijson with open("records.json", "rb") as f: parser = ijson.items(f, "records.item.payload") for payload in parser: handle(payload)这里的records.item.payload是JSONPath形式的过滤路径,只提取记录里的指定子字段。对于不需要全字段的统计任务,这种定向解析能省不少CPU和内存。不过ijson不是标准库,引入前要评估打包环境和部署成本。
如果不想引入第三方库,还有一个标准库思路:用json.JSONDecoder().raw_decode()配合自定义的流式读取。raw_decode()可以只解析满足完整JSON对象的最小前缀,并返回剩余字符串,这就允许我们从大字符串中逐个提取JSON对象。
import json decoder = json.JSONDecoder() pos = 0 text = '[{"a":1}, {"b":2}, {"c":3}]' while pos < len(text): obj, end = decoder.raw_decode(text[pos:]) print(obj) pos += end while pos < len(text) and text[pos].isspace(): pos += 1这个技巧在处理拼接的多个JSON对象文本时非常实用,但要注意raw_decode()不会像loads那样校验整个字符串的剩余部分,需要自己控制循环边界。理解原理后,你甚至可以封装一个轻量级流式解析器。
6.3 数据管道中批量读写的合并技巧
在批量处理场景中,一个常见的需求是把几千条记录合并存入一个JSON数组。很多人会写一个循环,每条记录循环一次json.dump(),结果生成的文件里格式是合法的,但性能很差。正确做法是先在内存中构造完整的列表,然后一次性json.dump()。
import json records = [] for record in fetch_from_db(): records.append(record) with open("export.json", "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False, indent=2)一次写入和多次写入的差距不仅仅在I/O次数,还在于json.dump()在序列化时会构建完整的输出字符串,多次调用会产生大量小字符串对象,GC压力也会上升。能一次写完就一次写完。
另一个合并技巧是使用标准库的json.JSONEncoder子类复用自己的类型转换逻辑,这样在批量序列化大数据时,比每次调用json.dump()传default函数更清爽,因为JSONEncoder可以复用同一套转换规则。
class CustomEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, Decimal): return float(obj) return super().default(obj) with open("out.json", "w", encoding="utf-8") as f: json.dump(data, f, cls=CustomEncoder, ensure_ascii=False)用过之后我发现,虽然只是把default函数封装到了类里,但代码组织更清晰,也不容易在多个地方写出不一致的转换逻辑。
7. 实战案例:一个带备份和校验的JSON配置管理模块
7.1 需求设定与结构设计
让前面所有内容落到一个具体项目上。假设我要做一个命令行工具,它需要一个配置文件来保存服务地址、超时时间、重试次数、是否开启调试模式等参数。工具启动时读取配置,命令行参数可以临时覆盖配置,退出前要把修改后的配置保存回文件。
这个需求看起来很常规,但实际隐含了几个难点:
- 配置文件可能不存在,程序第一次启动时要生成默认配置。
- 配置文件可能损坏,不能因为它坏了就导致整个程序无法启动。
- 用户可能修改了配置,修改后要尽可能防止写入覆盖失败。
- 数据里有中文,写入后不能变成
\uXXXX,否则不便于人工维护。 - 配置变更前最好有一个备份文件,便于回滚。
针对这些问题,我把模块划分成几个函数:默认配置生成、带备份的安全读取、原子写入、参数覆盖合并。
7.2 核心代码实现
先定义配置项的默认值和加载入口。
import json import os import shutil import tempfile from copy import deepcopy DEFAULT_CONFIG = { "server": { "host": "127.0.0.1", "port": 8080, }, "runtime": { "timeout": 5, "retries": 3, "debug": False, }, "message": "默认提示", } CONFIG_PATH = "app_config.json" BACKUP_PATH = "app_config.json.bak" def load_config(): if not os.path.exists(CONFIG_PATH): current = deepcopy(DEFAULT_CONFIG) save_config(current) return current try: with open(CONFIG_PATH, "r", encoding="utf-8-sig") as f: current = json.load(f) except (json.JSONDecodeError, UnicodeDecodeError) as e: if os.path.exists(BACKUP_PATH): with open(BACKUP_PATH, "r", encoding="utf-8-sig") as f: current = json.load(f) else: current = deepcopy(DEFAULT_CONFIG) save_config(current) return current def save_config(config): tmp_path = CONFIG_PATH + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(config, f, indent=2, ensure_ascii=False) f.flush() os.fsync(f.fileno()) os.replace(tmp_path, CONFIG_PATH) # 每次保存后同步生成备份文件 shutil.copy2(CONFIG_PATH, BACKUP_PATH) def merge_cli_overrides(config, cli_args): merged = deepcopy(config) if cli_args.get("host"): merged["server"]["host"] = cli_args["host"] if cli_args.get("port"): merged["server"]["port"] = cli_args["port"] return merged这里有几个设计上的细节值得说明。encoding="utf-8-sig"在读取时会自动剥离BOM头,即使文件来自Windows环境也能正确解析。os.fsync(f.fileno())会把缓冲区数据强制持久化到磁盘,避免操作系统还没落盘程序就返回成功。这个操作对配置文件这类关键数据非常重要,虽然会慢一点点,但可靠性优先。
备份文件的生成方式用的是shutil.copy2,它会保留文件的权限和时间戳信息。每次保存后都同步更新备份,这样配置一旦损坏,恢复的备份不会太旧。
7.3 边界情况测试与踩坑记录
在实际使用这个模块时,我遇到过一个值得记录的边界场景:配置文件写入过程中,程序在json.dump()之后、os.replace()之前被杀,这时app_config.json.tmp残留下来,而正式的配置文件还是旧内容。下次程序启动时读取旧配置,一切正常,但磁盘上多了一个没被清理的临时文件。如果临时文件不清理,日积月累会占用空间。
解决方法是启动时检查并清理同目录下的.tmp文件。如果你要做得更严谨,启动加载配置时还可以先校验配置中是否有必要字段,缺少关键字段时要补充默认值,而不是直接覆盖。比如用户手动删掉了runtime段,程序要能感知并自动恢复默认。
def ensure_defaults(config): changed = False for key, default_value in DEFAULT_CONFIG.items(): if key not in config: config[key] = deepcopy(default_value) changed = True return changed这些边界情况是测试驱动的重点。我习惯在本地准备几个测试数据文件:空文件、损坏文件、带BOM的文件、缺少字段的文件、超深层嵌套文件,逐一跑一遍加载逻辑,确保不会出现难以理解的崩溃。
7.4 从配置文件模块扩展到通用JSON工具
上面这套配置管理模块在逻辑上其实可以泛化。你只需要把“配置”换成“任意结构化数据”,就能变成一种通用的JSON存档工具。我在另一个项目里就做过类似的事情:用户可以在界面里编辑一组分组规则,规则保存为一个不断积累的JSON列表,每次保存都走同样的原子写入和备份路径,效果非常稳定。
如果后续想扩展到更复杂的场景,可以在这套模块基础上增加JSON Schema校验。配置结构复杂到一定程度后,用jsonschema库做字段校验能提前发现很多类型错误。
import jsonschema schema = { "type": "object", "properties": { "server": { "type": "object", "properties": { "host": {"type": "string"}, "port": {"type": "integer", "minimum": 1, "maximum": 65535} }, "required": ["host", "port"] } }, "required": ["server", "runtime"] } jsonschema.validate(config, schema)我用过很多次jsonschema之后,最大的体会是校验逻辑和业务代码分离,让维护成本低了很多。这个扩展方向非常适合配置越来越复杂、错误反馈要求越来越清晰的系统。
写JSON文件读写这件事,底层逻辑始终是“输入可控、过程可停、结果可预测”。我在实操中最大的体会是:真正出问题的场景往往不是API不会用,而是对文件编码、数据结构和异常边界没有足够敬畏。每多掌握一个细节,线上就能少排查半天问题。如果你也在做类似的数据存储模块,建议先把你手头最常发生的三种异常写进自动处理逻辑,再逐步补充健壮性设计。这样下来,一套稳定的JSON读写模块并不难落地。