资讯详情 从字节流到哈希表:用状态机实现一个极简JSON解析器
📅 2026/10/17 15:15:29
上个月接了个数据清洗的小工具日志里塞了一堆客户端上报的 JSON我图省事直接上正则抠age: 18这类字段结果嵌套层级一深就全乱了字符串里混着大括号和反斜杠的场景更是直接翻车。后来老老实实把解析器的原理捋了一遍写了个极简实现才算真正看懂“解析器像是个数据翻译官”这句话的分量字节流本身只是一串连续字符是解析器按语法规则把它翻译成程序能直接操作的内存结构{age: 18}才变成了哈希表在 Python 里就是字典。这篇文章就用一个极简状态机实现来拆解全过程适合所有想弄懂编译原理入门、或者被字符串解析反复折磨过的开发者。1. 解析器到底在做什么从字节流到内存结构中间多了一层语法1.1 数据翻译官的本职词法与语法两顶帽子很多人第一次接触解析器容易把它想得很玄。其实翻译官的工作可以拆成两件事先把连续字符切成有意义的“单词”再把这些单词按规则组装成结构。切单词这一步叫词法分析。JSON 的“单词”类型不多左花括号、右花括号、左方括号、右方括号、冒号、逗号、字符串、数字、true、false、null。每个单词都有自己的边界比如字符串必须以双引号包裹数字必须符合 JSON 的数字语法。解析器逐个读字符遇到空白就跳过遇到就知道要进入字符串状态遇到数字或负号就知道要读一个数字。组装这一步叫语法分析。{age: 18}不是几个单词随便堆在一起对象必须是“键、冒号、值、逗号或右花括号”的固定节奏值又可以是对象和数组于是嵌套结构就出现了。语法分析负责把这个节奏校验一遍同时把值塞进字典里。极简实现里词法和语法往往交织在同一个函数中但思考问题的时候一定要把它们分开后面调 bug 会轻松很多。1.2 从 {age: 18} 出发的一条解析路径我们跟着{age: 18}这串字符走一遍看看翻译官是怎么处理它的步骤当前字符解析器判断累积结果1{进入对象模式期待一个键空字典2age读到字符串作为键等待冒号3:键值分隔符期待值等待值418读到整数{age: 18}5}对象结束返回字典每一步解析器只关心两件事当前字符是什么当前处于什么语法状态。字符走到第 4 步时解析器不会把18直接塞进某个字符串里而是先把它转换成整数再绑定到键age上。这个转换动作就是“数据翻译”最直观的体现。1.3 目标数据结构语言里的“哈希表”标题里说“变成内存里的哈希表”这里得补充一句哈希表是底层实现不同语言的叫法不一样。Python 里是dictJava 里是HashMapC 里是unordered_map。它们的共同点是键值对存储、平均复杂度 O(1) 查找。JS 的Object在语义上也扮演类似角色但底层实现细节和哈希表并不完全一致。理解这一点对设计解析器很重要解析器输出的不是文本而是目标语言里真正可用的数据结构。所以parse_object创建字典、parse_array创建列表、parse_string返回字符串、parse_number返回整数或浮点数。这个“从语法到运行时对象”的映射是解析器与普通文本处理工具最本质的区别。2. 极简状态机的骨架一个字符一个字符喂给解析器2.1 状态定义与转移逻辑状态机不是非得画一张密密麻麻的状态图。对于 JSON 这种嵌套结构最简单的做法是用一个parse_value函数做总调度根据当前字符分发到不同的解析函数每个解析函数内部再通过循环和条件分支维护自己的状态。先列出极简版的核心状态转移关系当前状态触发字符动作下一状态START初始空格、换行等空白跳过STARTSTART{调用对象解析对象内部START[调用数组解析数组内部START调用字符串解析字符串结束START数字或-调用数字解析值结束STARTt/f/n调用字面量解析值结束STRING普通字符收集字符STRINGSTRING\进入转义处理转义子状态STRING结束字符串返回调用方这张表看起来简单但它覆盖了解析器 90% 的字符处理逻辑。剩下的对象和数组内部细节只是在调用完parse_value之后再检查“逗号还是右括号”这两个标点而已。2.2 parse_value整台机器的总调度台我把核心代码拆开讲结构是这样class TinyJSONParser: def __init__(self, text): self.text text self.pos 0 def skip_ws(self): while self.pos len(self.text) and self.text[self.pos] in \t\n\r: self.pos 1 def parse_value(self): self.skip_ws() if self.pos len(self.text): raise ValueError(unexpected end of input) ch self.text[self.pos] if ch {: return self.parse_object() if ch [: return self.parse_array() if ch : return self.parse_string() if ch t: return self.parse_literal(true, True) if ch f: return self.parse_literal(false, False) if ch n: return self.parse_literal(null, None) if ch - or ch.isdigit(): return self.parse_number() raise ValueError(funexpected char {ch!r} at {self.pos})parse_value是整个解析器的心脏。它的设计意图很明确先把所有空白吃掉然后看第一个有效字符是什么再决定调用哪个子函数。因为 JSON 的每个“值”都可能是对象、数组、字符串、数字或字面量所以所有嵌套结构的入口都收拢在这一处。数组里的每个元素要调用parse_value对象里的每个值也要调用parse_value这样一个函数就把递归的骨架撑起来了。2.3 parse_string转义处理里的隐性状态字符串解析是状态机最密集的地方因为引号内部还有一层“转义”状态def parse_string(self): if self.text[self.pos] ! : raise ValueError(expect string starts with quote) self.pos 1 out [] while self.pos len(self.text): ch self.text[self.pos] if ch : self.pos 1 return .join(out) if ch \\: self.pos 1 if self.pos len(self.text): raise ValueError(incomplete escape) esc self.text[self.pos] mapping { : , \\: \\, /: /, b: \b, f: \f, n: \n, r: \r, t: \t, } if esc in mapping: out.append(mapping[esc]) self.pos 1 elif esc u: hex_part self.text[self.pos1:self.pos5] if len(hex_part) 4 or any(c not in 0123456789abcdefABCDEF for c in hex_part): raise ValueError(invalid \\u escape) out.append(chr(int(hex_part, 16))) self.pos 5 else: raise ValueError(finvalid escape {esc!r}) else: if ord(ch) 0x20: raise ValueError(unescaped control character in string) out.append(ch) self.pos 1 raise ValueError(unterminated string)这里最关键的是转义分支遇到\时向后多看一个字符查表得到真实字符比如\n会被映射成换行符\会被映射成双引号。为什么不能直接把反斜杠原样保留因为 JSON 文本里的\n是两个字符而内存字符串里的换行是一个字符翻译官必须做这个转换。这是整台机器里最典型的“状态嵌套”场景外层是字符串状态内层是转义判断状态两层状态交叉推进。3. 递归组装对象与数组的嵌套结构如何长出来3.1 parse_object键、冒号、值、逗号的四拍节奏对象解析的规律很固定先跳过左花括号如果下一个字符是右花括号返回空字典否则循环执行“读键、读冒号、读值、读逗号或右花括号”四步。代码写成这样def parse_object(self): self.pos 1 # 跳过 { obj {} self.skip_ws() if self.pos len(self.text) and self.text[self.pos] }: self.pos 1 return obj while True: self.skip_ws() if self.pos len(self.text) or self.text[self.pos] ! : raise ValueError(object key must be a string) key self.parse_string() self.skip_ws() if self.pos len(self.text) or self.text[self.pos] ! :: raise ValueError(expect : after object key) self.pos 1 value self.parse_value() obj[key] value self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ,: self.pos 1 continue if self.pos len(self.text) and self.text[self.pos] }: self.pos 1 return obj raise ValueError(expect , or } in object)注意两个细节。第一键必须是字符串JSON 标准里不允许裸键名所以看到{age: 18}要直接报错。第二值的位置调用的是parse_value不是又写一个parse_string这就是递归的起点对象里可以嵌套对象值可以是任意 JSON 类型。3.2 parse_array 与递归的力量数组解析比对象还简单它没有“键”这个环节直接循环调用parse_valuedef parse_array(self): self.pos 1 # 跳过 [ arr [] self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ]: self.pos 1 return arr while True: value self.parse_value() arr.append(value) self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ,: self.pos 1 continue if self.pos len(self.text) and self.text[self.pos] ]: self.pos 1 return arr raise ValueError(expect , or ] in array)递归的力量在于parse_array里遇到[时又会调用parse_value而parse_value又可能回到parse_array。嵌套多深调用栈就跟多深不用手写任何复杂的状态栈调用栈本身就是语法树生长的地方。这也是为什么我在 2.1 说纯显式状态表不适合完整 JSON 语法——一旦出现嵌套上下文显式状态数量就会爆炸而递归下降天然把“当前上下文”保存在调用栈里代码量小且直观。3.3 完整跑通实验{age: 18} 最终长什么样把前面所有片段拼到一起就是一个完整可运行的极简 JSON 解析器。数字解析部分我用一行正则匹配JSON 数字语法是固定的“可选负号、整数、可选小数、可选指数”import re class TinyJSONParser: def __init__(self, text): self.text text self.pos 0 def skip_ws(self): while self.pos len(self.text) and self.text[self.pos] in \t\n\r: self.pos 1 def parse_value(self): self.skip_ws() if self.pos len(self.text): raise ValueError(unexpected end of input) ch self.text[self.pos] if ch {: return self.parse_object() if ch [: return self.parse_array() if ch : return self.parse_string() if ch t: return self.parse_literal(true, True) if ch f: return self.parse_literal(false, False) if ch n: return self.parse_literal(null, None) if ch - or ch.isdigit(): return self.parse_number() raise ValueError(funexpected char {ch!r} at {self.pos}) def parse_literal(self, word, value): if self.text.startswith(word, self.pos): self.pos len(word) return value raise ValueError(finvalid literal {word!r} at {self.pos}) def parse_string(self): if self.text[self.pos] ! : raise ValueError(expect string starts with quote) self.pos 1 out [] while self.pos len(self.text): ch self.text[self.pos] if ch : self.pos 1 return .join(out) if ch \\: self.pos 1 if self.pos len(self.text): raise ValueError(incomplete escape) esc self.text[self.pos] mapping {: , \\: \\, /: /, b: \b, f: \f, n: \n, r: \r, t: \t} if esc in mapping: out.append(mapping[esc]) self.pos 1 elif esc u: hex_part self.text[self.pos1:self.pos5] if len(hex_part) 4 or any(c not in 0123456789abcdefABCDEF for c in hex_part): raise ValueError(invalid \\u escape) out.append(chr(int(hex_part, 16))) self.pos 5 else: raise ValueError(finvalid escape {esc!r}) else: if ord(ch) 0x20: raise ValueError(unescaped control character in string) out.append(ch) self.pos 1 raise ValueError(unterminated string) def parse_number(self): pattern re.compile(r-?(?:0|[1-9]\d*)(?:\.\d)?(?:[eE][-]?\d)?) match pattern.match(self.text, self.pos) if not match: raise ValueError(finvalid number at {self.pos}) num_str match.group() self.pos match.end() if . in num_str or e in num_str or E in num_str: return float(num_str) return int(num_str) def parse_object(self): self.pos 1 obj {} self.skip_ws() if self.pos len(self.text) and self.text[self.pos] }: self.pos 1 return obj while True: self.skip_ws() if self.pos len(self.text) or self.text[self.pos] ! : raise ValueError(object key must be a string) key self.parse_string() self.skip_ws() if self.pos len(self.text) or self.text[self.pos] ! :: raise ValueError(expect : after object key) self.pos 1 value self.parse_value() obj[key] value self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ,: self.pos 1 continue if self.pos len(self.text) and self.text[self.pos] }: self.pos 1 return obj raise ValueError(expect , or } in object) def parse_array(self): self.pos 1 arr [] self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ]: self.pos 1 return arr while True: value self.parse_value() arr.append(value) self.skip_ws() if self.pos len(self.text) and self.text[self.pos] ,: self.pos 1 continue if self.pos len(self.text) and self.text[self.pos] ]: self.pos 1 return arr raise ValueError(expect , or ] in array) def parse(self): result self.parse_value() self.skip_ws() if self.pos ! len(self.text): raise ValueError(ftrailing data at {self.pos}) return result运行parser TinyJSONParser({age: 18}) print(parser.parse())输出{age: 18}{age: 18}这 12 个字节走完所有状态后最终落进了一个 Python 字典。可以打印类型确认result parser.parse() print(type(result)) # class dict print(type(result[age])) # class int这就是“从字节流到哈希表”的直接证据。4. 边界与异常JSON解析最容易翻车的五个细节4.1 代理对与 \u 转义乱码的源头刚才的parse_string里遇到\u会直接chr(int(hex_part, 16))这个写法能处理基本 BMP 字符但处理不了 emoji 这类需要代理对的字符。JSON 里一个 emoji 经常被编码成两个转义序列\ud83d\ude00它们必须被合并成一个 Unicode 码位再转换成对应字符。如果只把两个代理项分别转成字符结果就是两个乱码符号。正确做法是先读第一个\uXXXX判断它是否落在高代理区间0xD800 到 0xDBFF如果是继续读下一个\uXXXX验证是低代理区间后用公式(high - 0xD800) * 0x400 (low - 0xDC00) 0x10000合成最终码位。这一步不看规范文档基本写不对很多底层解析库的 bug 都出在这。4.2 数字格式的严格性前导零和加号都非法JSON 数字有一套严格语法整数部分不能以零开头除非这个数就是0不允许出现5小数点后面必须跟数字。我的正则-?(?:0|[1-9]\d*)(?:\.\d)?(?:[eE][-]?\d)?已经把这些约束写进去了但如果你是手写循环判断数字很容易踩这些坑。还有个隐蔽问题大整数超出目标语言安全整数范围怎么办Python 的int任意精度没问题但如果解析到 JS 或者对精度敏感的系统1e999这类超大数字会变成inf或直接溢出。工程上往往是先按字符串保存再由调用方决定用什么类型承接。4.3 未闭合字符串与非法控制字符双层引号包裹的字符串里如果文件在结尾处突然断了解析器必须能报“unterminated string”而不是继续循环直到数组越界。我的实现对越界做了检查但很多初学者写的版本会在while循环里悄悄访问self.text[self.pos]越界。另一个容易忽略的是控制字符。JSON 标准要求字符串内不允许直接出现未转义的\n、\t这类 ASCII 控制字符它们必须写成\n、\t的形式。原因是原始字符没法区分“换行在字符串里”还是“字符串外的换行”直接引入会让文本结构变模糊。实用的检查就是ord(ch) 0x20就拒绝。4.4 尾随数据/尾随逗号/注释宽容模式要不要开parse()最后一行检查剩余字符这个动作能挡住大量异常输入。比如{a: 1} extra如果不检查解析器会把extra当作不存在而静默通过这在日志处理里非常危险因为你以为拿到了完整对象其实后面还滚了一大串脏数据。尾随逗号和注释则是另一个相反的问题标准 JSON 不允许但很多配置文件和 API 返回却爱这么写。工程里常见的做法是加一个“宽容模式”开关比如allow_trailing_commaTrue时对象解析遇到{,}或[,]就跳过注释同理。但这个开关默认必须关闭否则你在严格协议下解析时会意外接受一堆非法输入埋下兼容性隐患。4.5 深层嵌套与栈溢出设置深度上限递归下降最怕的就是深度攻击或意外坏数据。构造一万层嵌套的[[[[...]]]]会让 Python 抛RecursionError更糟的情况是程序直接崩掉。标准库里很多解析器都有递归深度上限默认几十层到上百层。极简版加一个计数器就行parse_value进入时depth 1超过阈值直接报错对象和数组解析完毕后depth - 1。这里有个取舍深度上限太低会拒绝合法但很深的业务数据太高又挡不住栈溢出。我通常默认 512 层起步具体值要结合调用栈大小来调而不是拍脑袋给一个数。5. 从教学原型到工程级解析器性能、扩展与正确性经验5.1 性能差距到底在哪极简版能跑通功能但性能离工程级还差很远。最明显的问题是逐字符 Python 循环每一个字符都要做类型判断、索引访问、分支跳转开销远高于用 C 实现的标准 JSON 模块。实测解析一个几十 MB 的 JSON 文件纯 Python 版可能慢两个数量级差距主要不在算法而在解释器层级的每字符开销。想优化第一是合并字符扫描比如跳过空白时一次跳一整段第二是字符串拼接不要一个个字符用列表收集再join这也是我上面代码的做法第三是数字解析直接用正则而不是手写逐位判断正则底层是匹配引擎批量处理的比手动循环快。真到生产环境还是优先用语言自带的标准 JSON 库或成熟第三方实现但读懂原理之后再去调优它们你会知道问题出在哪儿而不是瞎试。5.2 扩展方向增量解析、AST 与方言模式教学版是一次性读完整个字符串再解析碰上超大文件几个 GB 的日志流就不合适了。工程上可以改成流式解析边读边跳过空白边解析边释放内存不需要一次性构建整个结构。另一个方向是输出 AST抽象语法树。普通解析直接构建字典和列表会丢掉位置信息、键值顺序细节和原始格式没法做精准的 diff、格式化或代码生成。AST 模式会为每个 token 记录start、end位置下一步想做 JSON 转换器或者 JSON 恢复器把数值类型还原子成原始字符串时这些位置信息非常重要。方言模式前面提过主要就是尾随逗号、注释、单引号字符串、裸键名。每支持一种方言都必须加独立开关别一股脑全塞进去否则你的“JSON 解析器”就变成了“JSON 超集解析器”别人拿着它解析标准 JSON 时反而会被宽容行为掩盖掉错误数据。5.3 测试策略用“语料库 模糊输入”兜住正确性写解析器不像写业务代码不能只测一个 happy path 就完事。我的做法是准备三批用例第一批是标准用例合法的各种嵌套、转义、数字格式都覆盖一遍第二批是非法用例未闭合字符串、尾随逗号、非法数字、多余数据都得有第三批是随机模糊输入跑几千次不崩溃、不出现越界访问是底线。对照标准行为还有一个技巧测试解析结果时用语言自带的标准库解析同样的字符串再递归比较两个结果结构是否完全一致。这能帮你快速发现转义、数字类型、嵌套处理的小偏差比肉眼盯代码高效得多。我自己在写这类解析器时最大的体会是真正麻烦的从来不是主流程而是每个分支的错误路径。状态机的骨架 20 分钟就能搭完调试那些pos偏移、代理对、尾随数据却要花上一整天。如果你也想练手建议先把“每个报错信息里带位置下标”这个习惯养成调试时能少走无数弯路。最后一个小技巧开发时在parse_value入口加一行print(self.text[self.pos], self.pos)开着这个开关跑一遍测试集你会亲眼看到那串字节是如何一步步被翻译成内存结构的。