放弃JSON吧！这4种数据格式让API性能提升5倍

admin

2026年1月30日 20:6 本文热度 693

我直接说结论啊：JSON 当然能用，但是，真别啥都往 JSON 上怼了，很多接口纯纯是被 JSON 拖慢的。

这个事是前两天晚上一点多，我在公司楼下啃着还剩半口的汉堡，运维给我打电话，说：“东哥，接口又 99% CPU 了，你看看是不是要不要先扩一波机器？” 我一听就知道，十有八九，又是我们那个“全家桶 JSON 接口”在作妖。

那个接口干嘛的呢？一个请求要查 3 个库、拼一堆对象，然后 json.dumps 一发，响应体三四百 KB 往外甩。用户量一上去，网络带宽、CPU、GC，全线拉满。那种感觉，跟你拿 Excel 开个 3G 的 CSV 差不多，人是清醒的，电脑已经在原地冻结了。

后来我们把那条链路的 JSON 全部换成二进制格式，压测下来，QPS 提升了差不多 4～6 倍，延迟直接砍成一半都不止，这才敢跟老板说一句“问题基本搞定了”。

所以我今天就当在群里跟你唠嗑，说下我现在做内部 API，常用的 4 个数据格式，都是 Python 里可以直接上的那种，不是啥玄学玩意。

你先别急着说“JSON 不挺好吗”

你回想一下你这些年被 JSON 坑过几次：

明明只是改了个字段名，结果一堆客户端挂掉
有人随手 datetime 弄成字符串，有人弄成时间戳，然后各种 KeyError、TypeError
日志里打印一坨 {...}，出了问题你只知道“是个对象”，具体里头啥值，得一点点数

更狠的是性能。之前我在做一个高并发系统的时候，数据库压测下来还挺稳的，换成 Postgres 以后吞吐直接翻倍，结果接口还是慢，最后发现瓶颈根本不在数据库，在序列化反序列化上，跟当年对比 Postgres 和 MySQL 性能的时候那个感觉很像。

而且 JSON 有一些天然的缺点：它是文本，体积大；它没 schema，类型全靠心照不宣；它解析是通用 parser，没法像特定协议那样“见字节就知道怎么走”。

所以合理的姿势不是“把 JSON 砸了”，而是：外部开放接口你爱用就用，没问题；内部系统、微服务之间，尤其是高 QPS 的链路，能换的地方，都值得考虑换。

我就按难度从低到高说 4 个选项，你可以照着自己的项目选。

先来个“没啥心智负担”的：MessagePack

这个真的是，几乎等价于“二进制版 JSON”。

你把它想成：结构还是 dict / list / int / str 这一套，就是不再转成 "{}" 这种文本，而是压成紧凑的二进制，解析也更快。

我们原来一个 Python 服务里，大量是这种结构：

import json

payload = {
    "user_id": 123,
    "nickname": "东哥",
    "tags": ["python", "backend", "打杂"],
    "profile": {
        "age": 30,
        "city": "Shanghai",
    }
}

data = json.dumps(payload).encode("utf-8")
print("json size =", len(data))

你稍微改两行就能上 MessagePack：

import msgpack  # pip install msgpack

payload = {
    "user_id": 123,
    "nickname": "东哥",
    "tags": ["python", "backend", "打杂"],
    "profile": {
        "age": 30,
        "city": "Shanghai",
    }
}

data = msgpack.packb(payload, use_bin_type=True)
print("msgpack size =", len(data))

# 服务端解析：
obj = msgpack.unpackb(data, raw=False)
print(obj["nickname"])

当时我在本机简单压了一下，一样的 payload，JSON 大概 400 多字节，MessagePack 能缩到两百左右。真实线上场景里，字段再多一点，嵌套复杂一点，节省的带宽就更明显了。

你想象一下： QPS 1w，单个响应体从 400KB 变成 80KB，网络 IO 直接降了 5 倍，你说整体性能能不涨么。

更现实一点的用法是，在内部接口里直接让 FastAPI / Flask 返回 MessagePack：

from fastapi import FastAPI, Response
import msgpack

app = FastAPI()

@app.get("/internal/user")
def get_user():
    # 假装这是查库查出来的
    payload = {"id": 1, "name": "dong", "score": 99}
    body = msgpack.packb(payload, use_bin_type=True)
    return Response(
        content=body,
        media_type="application/x-msgpack"
    )

客户端如果是 Python，也很简单：

import requests, msgpack

resp = requests.get("http://svc/internal/user")
data = msgpack.unpackb(resp.content, raw=False)
print(data["score"])

这里最香的一点就是：迁移成本极低，数据结构不动，只是换个“打包方式”，而且你想做双写、灰度都很容易，某些链路只对接支持的人用。

接下来有点“正规军味道”的：Protocol Buffers（Protobuf）

Protobuf 就是那种，你一看就知道是大厂爱用的玩意。

它的套路跟 JSON 完全不一样： “先写 schema，再生成代码，再用代码读写”。

比如有个用户资料接口，我们先写个 .proto 文件（随便起名 user.proto）：

syntax = "proto3";

message UserProfile {
  int64 id = 1;
  string name = 2;
  int32 age = 3;
  string city = 4;
  repeated string tags = 5;
}

然后你用官方工具生成 Python 代码（这步一般是 CI 里跑，自己电脑敲一下就懂了）：

protoc --python_out=. user.proto

会生成一个 user_pb2.py，里面有 UserProfile 这个类，之后写代码就变成这样：

from user_pb2 import UserProfile

# 序列化
user = UserProfile(
    id=123,
    name="dong",
    age=30,
    city="Shanghai",
    tags=["python", "backend"],
)
data = user.SerializeToString()
print("protobuf size =", len(data))

# 反序列化
user2 = UserProfile()
user2.ParseFromString(data)
print(user2.name, user2.tags)

为啥这个东西性能好？

有 schema，字段类型、编号都提前规划好了
底层按“字段号 + 类型 + 值”那套规则贴着字节走
不需要解析字符串、走 AST 之类的慢路径

我之前做一个服务间调用，把 requests + JSON 换成 gRPC + Protobuf，压测下来单次调用延迟从 25ms 掉到 6～7ms，QPS 差不多翻了 3 倍。再叠加连接复用之类的优化，整条链路拉到 4～5 倍确实不夸张。

当然它的代价也明显：改字段要考虑兼容性，字段号不要乱动；调试不如 JSON 直观，你要么写个小工具把二进制 decode 出来，要么配合 gRPC UI 之类的看。

所以我现在的习惯是：

对外 HTTP API：还是 JSON，方便第三方调，毕竟人家不想为了调你接口先编译个 .proto
内部微服务 / 大流量链路：优先 Protobuf 或 MessagePack，看团队习惯

你会发现，这种有 schema 的协议，一旦你用习惯了，很多“类型错一半天才报”的 BUG 都消失了，跟当年改 Spring Boot 默认配置，不再被一些奇怪的默认值坑一样，是那种“用了才知道爽”的东西。

第三个我挺喜欢拿来做“数据管道”的：Avro

Avro 跟 Protobuf 有点像，也是 schema 驱动，但它的 schema 本身就是 JSON，这点在做大数据那一挂的时候特别受欢迎。

你看一个简单的 Avro schema（写成 Python 里的字典也行）：

user_schema = {
    "type": "record",
    "name": "UserEvent",
    "fields": [
        {"name": "id", "type": "long"},
        {"name": "event", "type": "string"},
        {"name": "ts", "type": "long"},
    ],
}

然后你用 fastavro 之类的库（pip install fastavro）可以这样打包一条记录：

from io import BytesIO
from fastavro import schemaless_writer, schemaless_reader

def encode_user_event(record):
    buf = BytesIO()
    schemaless_writer(buf, user_schema, record)
    return buf.getvalue()

def decode_user_event(data):
    buf = BytesIO(data)
    return schemaless_reader(buf, user_schema)

event = {
    "id": 123,
    "event": "LOGIN",
    "ts": 1736760000,
}

blob = encode_user_event(event)
print("avro size =", len(blob))
print(decode_user_event(blob))

那它为啥跟 API 性能有啥关系？

我们有一条链路是这样的： HTTP 进来 -> 校验 -> 打包成事件 -> 丢 MQ / Kafka -> 下游服务慢慢处理。

HTTP 这一层你可以还是 JSON，但在 MQ 里的那坨数据，强烈建议用 Avro/Protobuf 这种二进制格式。

你看之前那篇 Kafka 不丢消息的文章，里面提到的一堆性能点，其实都跟消息大小、解析速度有关。

同样一秒几万条消息，JSON 每条 2KB，Avro 压到 500B， Broker 压力、网络、消费者解析 CPU，都能省一大截。

而且 Avro 的 schema 演进特别友好：加字段可以有默认值；删字段下游可以忽略；这对那种“永远在加字段”的埋点、日志、事件系统太友好了。

第四个我会留给“极端场景”的：CBOR 或 FlatBuffers

这个就看你项目是什么类型了，我简单说两种。

如果你做的是 IoT、嵌入式、小设备， CBOR 会是一个挺自然的选择，它跟 JSON 的关系，大概就是“更强的 MessagePack”：

支持更多类型：有二进制、日期、bigint、tag 等等
设计本身就是“适合受限设备”的标准

Python 这边用起来也不麻烦：

import cbor2  # pip install cbor2

payload = {
    "t": 23.5,
    "h": 0.45,
    "ts": 1736760123,
}

data = cbor2.dumps(payload)
print("cbor size =", len(data))

obj = cbor2.loads(data)
print(obj["t"], obj["h"])

当时我们搞一个小设备上报状态的接口，一开始偷懒直接 HTTP + JSON，结果设备那边说：

“兄弟，我们这破 ARM 板子 JSON 序列化 CPU 打满了。”

换成 CBOR 以后，带宽直接少了一半多一点，设备端用 C 解析 CBOR 也比 JSON 轻松很多，整个链路就顺滑了，不再出现那种“卡在 70% 不动”的离谱情况，跟我之前排查那个 1024 字节 TCP 包卡死的 BUG 的感觉类似——很多问题，本质都是“字节怎么组织”的事情。

如果你做的是游戏、客户端渲染、移动端那种， FlatBuffers / Cap’n Proto 这一类“零拷贝”的格式就很有意思：它的思路是“数据结构按内存布局写死在字节里，解析的时候不需要再 copy 一份出来”，这个在 Python 里用起来稍微有点重，我一般是 C++/Rust/Go 那边玩得多，就不展开了，你只要知道：这类协议可以把“解析时间”压得非常低，适合频繁传输、结构固定的东西。

那到底能不能“提升 5 倍”？

这个问题我当时也纠结过，因为你很难用一句话去概括所有场景，有时候只是 JSON 改成 MessagePack，网络带宽省个 40%，CPU 省个 20%，整体 QPS 涨个 1.5 倍，有时候是“HTTP + JSON”改成“gRPC + Protobuf + 长连接”，外加把数据库、线程池、连接池这些配置从默认值改一改，综合下来，整条链路性能翻个 4～6 倍也一点不夸张，就跟当年改 SpringBoot 默认配置，不改百分百踩坑一个道理。

如果你想自己在本地做个“小实验”，可以搞个最粗暴的脚本：

import json
import msgpack
import time
from dataclasses import dataclass, asdict

@dataclass
class User:
    id: int
    name: str
    email: str
    tags: list
    score: float

payload = [asdict(User(
    id=i,
    name=f"user-{i}",
    email=f"user{i}@test.com",
    tags=["python", "backend", "性能测试"],
    score=i * 0.1
)) for i in range(1000)]  # 一千条数据

def bench_encode(fn, label):
    t0 = time.perf_counter()
    for _ in range(200):
        fn()
    t1 = time.perf_counter()
    print(label, "cost =", (t1 - t0))

def json_encode():
    return json.dumps(payload).encode("utf-8")

def msgpack_encode():
    return msgpack.packb(payload, use_bin_type=True)

bench_encode(json_encode, "json")
bench_encode(msgpack_encode, "msgpack")

你随便跑一跑，大概率能看到 MessagePack 比 JSON 快一截、体积小一截，如果把 Protobuf/Avro 什么的也加进来，你会更直观地看到差距。

当然了，别指望“我把 JSON 换成 Protobuf，啥都不调，就从 1000 QPS 变成 5000 QPS”，系统是一个整体，数据库、线程池、连接池、磁盘 IO、网络延迟，每一环都得看。

怎么在现有项目里“动手不动脚”地换

我知道很多人一听到“换协议”就头皮发麻： “我们线上几十个服务，咋可能说换就换？”

实际上可以非常温和：

第一步，挑一条内部链路做试验田比如 A 服务调 B 服务，本来就是内部 RPC，你可以从“同时支持 JSON + 二进制”开始。

HTTP 里最简单的办法就是看 Accept 头：

from fastapi import FastAPI, Request, Response
import json, msgpack

app = FastAPI()

@app.post("/internal/do_something")
asyncdef do_something(req: Request):
    body = await req.body()
    # 为了简单起见，这里只写 JSON
    data = json.loads(body.decode("utf-8"))

    result = {"ok": True, "value": data.get("x", 0) * 2}

    accept = req.headers.get("accept", "")
    if"application/x-msgpack"in accept:
        return Response(
            content=msgpack.packb(result, use_bin_type=True),
            media_type="application/x-msgpack",
        )
    else:
        return Response(
            content=json.dumps(result).encode("utf-8"),
            media_type="application/json",
        )