第26章 案例3：数据序列化库

数据序列化是现代系统间通信的基础设施。本章通过一个完整的序列化库封装案例，展示如何使用 pybind11 封装各种序列化格式，特别关注错误处理和性能优化。

26.1 JSON 序列化

nlohmann/json 集成

json_wrapper.cpp

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <nlohmann/json.hpp>

namespace py = pybind11;
using json = nlohmann::json;

// Python dict -> JSON
json dict_to_json(const py::dict& d) {
    json j;

    for (auto [key, value] : d) {
        std::string key_str = py::str(key).cast<std::string>();

        if (py::isinstance<py::str>(value)) {
            j[key_str] = value.cast<std::string>();
        } else if (py::isinstance<py::int_>(value)) {
            j[key_str] = value.cast<int64_t>();
        } else if (py::isinstance<py::float_>(value)) {
            j[key_str] = value.cast<double>();
        } else if (py::isinstance<py::bool_>(value)) {
            j[key_str] = value.cast<bool>();
        } else if (py::isinstance<py::list>(value)) {
            j[key_str] = py::cast<std::vector<py::object>>(value);
        } else if (py::isinstance<py::dict>(value)) {
            j[key_str] = dict_to_json(value.cast<py::dict>());
        } else {
            j[key_str] = py::str(value).cast<std::string>();
        }
    }

    return j;
}

// JSON -> Python dict
py::dict json_to_dict(const json& j) {
    py::dict d;

    for (auto& [key, value] : j.items()) {
        switch (value.type()) {
            case json::value_t::string:
                d[key.c_str()] = value.get<std::string>();
                break;
            case json::value_t::number_integer:
                d[key.c_str()] = value.get<int64_t>();
                break;
            case json::value_t::number_unsigned:
                d[key.c_str()] = value.get<uint64_t>();
                break;
            case json::value_t::number_float:
                d[key.c_str()] = value.get<double>();
                break;
            case json::value_t::boolean:
                d[key.c_str()] = value.get<bool>();
                break;
            case json::value_t::array: {
                py::list lst;
                for (auto& item : value) {
                    lst.append(json_to_dict(item));
                }
                d[key.c_str()] = lst;
                break;
            }
            case json::value_t::object: {
                d[key.c_str()] = json_to_dict(value);
                break;
            }
            case json::value_t::null:
                d[key.c_str()] = py::none();
                break;
            default:
                d[key.c_str()] = py::none();
        }
    }

    return d;
}

// 序列化接口
std::string json_serialize(const py::dict& d) {
    try {
        json j = dict_to_json(d);
        return j.dump();
    } catch (const std::exception& e) {
        throw py::value_error(std::string("Serialization error: ") + e.what());
    }
}

// 反序列化接口
py::dict json_deserialize(const std::string& s) {
    try {
        json j = json::parse(s);
        return json_to_dict(j);
    } catch (const json::parse_error& e) {
        throw py::value_error(std::string("Parse error: ") + e.what());
    } catch (const std::exception& e) {
        throw py::runtime_error(std::string("Deserialization error: ") + e.what());
    }
}

错误处理策略

json_error_handling.cpp

#include <pybind11/pybind11.h>
#include <nlohmann/json.hpp>

namespace py = pybind11;
using json = nlohmann::json;

// 自定义异常类
class SerializationError : public std::runtime_error {
public:
    SerializationError(const std::string& msg, int line = -1, int column = -1)
        : std::runtime_error(msg), line_(line), column_(column) {}

    int line() const { return line_; }
    int column() const { return column_; }

private:
    int line_;
    int column_;
};

// 安全的解析函数
py::dict safe_parse(const std::string& json_str) {
    // 检查空输入
    if (json_str.empty()) {
        throw py::value_error("Empty input string");
    }

    // 检查是否包含BOM
    std::string_view view = json_str;
    if (view.starts_with("\xEF\xBB\xBF")) {
        view = view.substr(3);
    }

    try {
        json j = json::parse(view);
        return json_to_dict(j);
    } catch (json::parse_error& e) {
        SerializationError err(
            std::string("JSON parse error: ") + e.what(),
            static_cast<int>(e.line),
            static_cast<int>(e.byte)
        );
        throw py::value_error(err.what());
    }
}

// 验证 JSON 格式
bool is_valid_json(const std::string& s) {
    try {
        json::parse(s);
        return true;
    } catch (...) {
        return false;
    }
}

// 格式化输出
std::string pretty_print(const std::string& json_str, int indent = 2) {
    try {
        json j = json::parse(json_str);
        return j.dump(indent);
    } catch (const json::parse_error& e) {
        throw py::value_error(std::string("Invalid JSON: ") + e.what());
    }
}

Python 接口

from json_wrapper import (
    json_serialize, json_deserialize,
    safe_parse, is_valid_json, pretty_print
)

data = {
    "name": "Alice",
    "age": 30,
    "scores": [95, 87, 92],
    "metadata": {
        "city": "Beijing",
        "active": True
    }
}

json_str = json_serialize(data)
print(f"Serialized: {json_str}")

parsed = json_deserialize(json_str)
print(f"Parsed: {parsed}")
print(f"Name: {parsed['name']}")

json_with_error = '{"name": "Bob", "age": }'  # 错误JSON
try:
    result = safe_parse(json_with_error)
except ValueError as e:
    print(f"Caught error: {e}")

test_cases = [
    '{"valid": true}',
    '{"incomplete":',
    'null',
    ''
]

for test in test_cases:
    print(f"'{test}' valid: {is_valid_json(test)}")

26.2 Protocol Buffers 集成

Protobuf 定义与生成

addressbook.proto

syntax = "proto3";

package tutorial;

message Person {
    string name = 1;
    int32 id = 2;
    string email = 3;

    enum PhoneType {
        MOBILE = 0;
        HOME = 1;
        WORK = 2;
    }

    message PhoneNumber {
        string number = 1;
        PhoneType type = 2;
    }

    repeated PhoneNumber phones = 4;
}

message AddressBook {
    repeated Person people = 1;
}

pybind11 Protobuf 绑定

protobuf_wrapper.cpp

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <google/protobuf/map.h>
#include <addressbook.pb.h>

namespace py = pybind11;
using namespace tutorial;

// Person 包装类
py::dict person_to_dict(const Person& person) {
    py::dict d;
    d["name"] = person.name();
    d["id"] = person.id();
    d["email"] = person.email();

    py::list phones;
    for (const auto& phone : person.phones()) {
        py::dict p;
        p["number"] = phone.number();
        p["type"] = Person::PhoneType_Name(phone.type());
        phones.append(p);
    }
    d["phones"] = phones;

    return d;
}

Person dict_to_person(const py::dict& d) {
    Person person;

    if (d.contains("name")) {
        person.set_name(py::str(d["name"]));
    }
    if (d.contains("id")) {
        person.set_id(py::int_(d["id"]));
    }
    if (d.contains("email")) {
        person.set_email(py::str(d["email"]));
    }
    if (d.contains("phones")) {
        py::list phone_list = py::cast<py::list>(d["phones"]);
        for (py::handle p : phone_list) {
            py::dict phone_dict = py::cast<py::dict>(p);
            Person::PhoneNumber* phone = person.add_phones();

            if (phone_dict.contains("number")) {
                phone->set_number(py::str(phone_dict["number"]));
            }
            if (phone_dict.contains("type")) {
                std::string type_str = py::str(phone_dict["type"]);
                Person::PhoneType type;
                if (Person::PhoneType_Parse(type_str, &type)) {
                    phone->set_type(type);
                }
            }
        }
    }

    return person;
}

// AddressBook 操作
py::dict addressbook_to_dict(const AddressBook& book) {
    py::list people;
    for (const auto& person : book.people()) {
        people.append(person_to_dict(person));
    }
    py::dict result;
    result["people"] = people;
    return result;
}

AddressBook dict_to_addressbook(const py::dict& d) {
    AddressBook book;

    if (d.contains("people")) {
        py::list people_list = py::cast<py::list>(d["people"]);
        for (py::handle p : people_list) {
            Person* person = book.add_people();
            *person = dict_to_person(py::cast<py::dict>(p));
        }
    }

    return book;
}

// 序列化/反序列化
std::string serialize_addressbook(const AddressBook& book) {
    std::string result;
    if (!book.SerializeToString(&result)) {
        throw py::runtime_error("Failed to serialize AddressBook");
    }
    return result;
}

AddressBook deserialize_addressbook(const std::string& data) {
    AddressBook book;
    if (!book.ParseFromString(data)) {
        throw py::value_error("Failed to parse AddressBook from string");
    }
    return book;
}

Protobuf 模块定义

protobuf_module.cpp

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include "protobuf_wrapper.h"

namespace py = pybind11;

PYBIND11_MODULE(protobuf_wrapper, m) {
    m.doc() = R"pbdoc(
        Protocol Buffers 序列化模块

        示例:
            >>> from protobuf_wrapper import Person, AddressBook
            >>> person = Person({"name": "Alice", "id": 1, "email": "alice@example.com"})
            >>> book = AddressBook({"people": [person]})
            >>> data = serialize(book)
    )pbdoc";

    // Person 类
    py::class_<Person>(m, "Person")
        .def(py::init<>())
        .def(py::init(&dict_to_person))
        .def_property("name", &Person::name, &Person::set_name)
        .def_property("id", &Person::id, &Person::set_id)
        .def_property("email", &Person::email, &Person::set_email)
        .def_property_readonly("phones", &person_to_dict)
        .def("__repr__", [](const Person& p) {
            return py::str("Person(name='{}', id={}, email='{}')")
                .format(p.name(), p.id(), p.email());
        });

    // AddressBook 类
    py::class_<AddressBook>(m, "AddressBook")
        .def(py::init<>())
        .def(py::init(&dict_to_addressbook))
        .def_property_readonly("people", &addressbook_to_dict)
        .def("add_person", &dict_to_person)
        .def("__len__", [](const AddressBook& book) {
            return book.people_size();
        });

    // 序列化函数
    m.def("serialize", &serialize_addressbook,
          "序列化 AddressBook 为二进制字符串");
    m.def("deserialize", &deserialize_addressbook,
          "从二进制字符串反序列化 AddressBook");
    m.def("person_to_dict", &person_to_dict);
    m.def("dict_to_person", &dict_to_person);
}

26.3 高性能序列化

FlatBuffers 集成

flatbuffers_wrapper.cpp

#include <pybind11/pybind11.h>
#include <flatbuffers/flatbuffers.h>
#include <flatbuffers/reflection.h>
#include "monster_generated.h"  // 生成的代码

namespace py = pybind11;
using namespace MyGame::Sample;

// FlatBuffers 包装器
class FlatBufferBuilder {
public:
    FlatBufferBuilder() : builder_(1024) {}

    py::bytes finish() {
        builder_.Finish(CreateMonsterDirect(
            &builder_,
            80,        // HP
            300,       // Mana
            "MyMonster",
            nullptr,   // Inventory
            Color_Blue,
            nullptr,   // Weapons
            nullptr,   // Equipment
            Vec(1.0f, 2.0f, 3.0f)
        ));
        return py::bytes(reinterpret_cast<const char*>(builder_.GetBufferPointer()),
                         builder_.GetSize());
    }

    void Reset() {
        builder_.Clear();
    }

private:
    flatbuffers::FlatBufferBuilder builder_;
};

// 从 Python 数据创建 FlatBuffer
py::bytes create_monster(const py::dict& data) {
    flatbuffers::FlatBufferBuilder builder(1024);

    // 解析数据
    auto name = data.contains("name") ?
        py::str(data["name"]) : "Default";

    // 创建 Vector of inventory (uint8 values)
    std::vector<uint8_t> inventory;
    if (data.contains("inventory")) {
        py::list inv = py::cast<py::list>(data["inventory"]);
        for (py::handle item : inv) {
            inventory.push_back(py::int_(item));
        }
    }

    // Build Monster
    auto name_offset = builder.CreateString(name.cast<std::string>());
    auto inv_offset = builder.CreateVector(inventory);

    auto monster = CreateMonster(builder, 100, 150, name_offset, inv_offset,
                                  Color_Green, 0, 0, MyGame::Sample::Weapon_ANY, 0);

    builder.Finish(monster);

    return py::bytes(reinterpret_cast<const char*>(builder.GetBufferPointer()),
                     builder.GetSize());
}

// 解析 FlatBuffer
py::dict parse_monster(const py::bytes& data) {
    auto buf = data.cast<std::string>();
    flatbuffers::Verifier verifier(
        reinterpret_cast<const uint8_t*>(buf.data()),
        buf.size()
    );

    if (!VerifyMonsterBuffer(verifier)) {
        throw py::value_error("Invalid FlatBuffer data");
    }

    auto monster = GetMonster(buf.data());

    py::dict result;
    result["hp"] = monster->hp();
    result["mana"] = monster->mana();
    result["name"] = monster->name()->str();

    py::list inventory;
    if (monster->inventory()) {
        for (size_t i = 0; i < monster->inventory()->size(); ++i) {
            inventory.append(monster->inventory()->Get(i));
        }
    }
    result["inventory"] = inventory;

    if (monster->pos()) {
        py::dict pos;
        pos["x"] = monster->pos()->x();
        pos["y"] = monster->pos()->y();
        pos["z"] = monster->pos()->z();
        result["pos"] = pos;
    }

    return result;
}

Cap’n Proto 集成

capnp_wrapper.cpp

#include <pybind11/pybind11.h>
#include <capnp/serialize.h>
#include <kj/io.h>
#include "addressbook.capnp.h"

namespace py = pybind11;

py::bytes serialize_address_book(const py::dict& data) {
    capnp::MallocMessageBuilder builder;
    AddressBook::Builder book = builder.initRoot<AddressBook>();

    // 设置 people
    if (data.contains("people")) {
        py::list people = py::cast<py::list>(data["people"]);
        auto people_vec = book.initPeople(people.size());

        for (size_t i = 0; i < people.size(); ++i) {
            py::dict person = py::cast<py::dict>(people[i]);
            Person::Builder p = people_vec[i];

            if (person.contains("name")) {
                p.setName(py::str(person["name"]));
            }
            if (person.contains("id")) {
                p.setId(py::int_(person["id"]));
            }
        }
    }

    // 序列化为字节
    kj::VectorOutputStream output;
    capnp::writeMessage(output, builder.getMessage());

    auto& vec = output.getArray();
    return py::bytes(reinterpret_cast<const char*>(vec.begin()), vec.size());
}

py::dict deserialize_address_book(const py::bytes& data) {
    auto buf = data.cast<std::string>();

    kj::ArrayInputStream input(
        kj::ArrayPtr<const uint8_t>(
            reinterpret_cast<const uint8_t*>(buf.data()),
            buf.size()
        )
    );

    capnp::InputStreamMessageReader reader(input);
    AddressBook::Reader book = reader.getRoot<AddressBook>();

    py::dict result;
    py::list people;

    for (auto person : book.getPeople()) {
        py::dict p;
        p["name"] = person.getName();
        p["id"] = person.getId();
        p["email"] = person.getEmail();
        people.append(p);
    }

    result["people"] = people;
    return result;
}

性能对比

序列化格式	编码速度	解码速度	体积效率	跨语言	模式演进
JSON	慢	慢	差	好	需版本号
Protocol Buffers	快	快	很好	好	支持
FlatBuffers	非常快	最快	很好	好	支持
Cap’n Proto	最快	最快	很好	一般	支持

26.4 错误处理机制

异常设计模式

error_handling.cpp

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <exception>

namespace py = pybind11;

// 自定义异常层次结构
class SerializationException : public std::runtime_error {
public:
    enum class ErrorCode {
        InvalidFormat,
        TruncatedData,
        VersionMismatch,
        FieldMissing,
        TypeMismatch,
        OutOfRange
    };

    SerializationException(ErrorCode code, const std::string& msg)
        : std::runtime_error(msg), code_(code) {}

    ErrorCode code() const { return code_; }

    static ErrorCode from_string(const std::string& s) {
        if (s == "invalid_format") return ErrorCode::InvalidFormat;
        if (s == "truncated_data") return ErrorCode::TruncatedData;
        if (s == "version_mismatch") return ErrorCode::VersionMismatch;
        if (s == "field_missing") return ErrorCode::FieldMissing;
        if (s == "type_mismatch") return ErrorCode::TypeMismatch;
        if (s == "out_of_range") return ErrorCode::OutOfRange;
        return ErrorCode::InvalidFormat;
    }

private:
    ErrorCode code_;
};

// 异常转换函数
void translate_exception(std::exception_ptr p) {
    try {
        if (p) std::rethrow_exception(p);
    } catch (const SerializationException& e) {
        py::dict exc_dict;
        exc_dict["type"] = "SerializationException";
        exc_dict["code"] = static_cast<int>(e.code());
        exc_dict["message"] = e.what();
        PyErr_SetObject(PyExc_ValueError, py::cast(exc_dict).ptr());
        return;
    } catch (const std::out_of_range& e) {
        PyErr_Format(PyExc_IndexError, "Out of range: %s", e.what());
        return;
    } catch (const std::exception& e) {
        PyErr_Format(PyExc_RuntimeError, "%s", e.what());
        return;
    }
    PyErr_SetString(PyExc_RuntimeError, "Unknown exception");
}

// 安全执行宏
#define PYBIND11_TRY_PARSE(...) \
    try { \
        __VA_ARGS__; \
    } catch (...) { \
        translate_exception(std::current_exception()); \
    }

// 验证函数
void validate_data(const py::dict& data, const std::vector<std::string>& required_fields) {
    for (const auto& field : required_fields) {
        if (!data.contains(field)) {
            throw SerializationException(
                SerializationException::ErrorCode::FieldMissing,
                "Missing required field: " + field
            );
        }
    }
}

bool validate_json_string(const std::string& s) {
    if (s.empty()) return false;

    // 检查控制字符
    for (char c : s) {
        if (c < 0x20 && c != '\t' && c != '\n' && c != '\r') {
            return false;
        }
    }
    return true;
}

Python 错误处理

from serializer import (
    serialize, deserialize,
    SerializationException, validate_data
)

class ValidationError(Exception):
    """Validation error with details"""
    def __init__(self, message, field=None, value=None):
        super().__init__(message)
        self.field = field
        self.value = value

def safe_deserialize(data, schema):
    """安全的反序列化，携带错误信息"""
    try:
        result = deserialize(data)

        # 验证结果
        required_fields = schema.get('required', [])
        missing = [f for f in required_fields if f not in result]

        if missing:
            raise ValidationError(
                f"Missing required fields: {missing}",
                field=missing
            )

        return result

    except SerializationException as e:
        # 解析异常代码
        error_codes = {
            0: 'invalid_format',
            1: 'truncated_data',
            2: 'version_mismatch',
            3: 'field_missing',
            4: 'type_mismatch',
            5: 'out_of_range'
        }
        print(f"Serialization error [{error_codes.get(e.code, 'unknown')}]: {e.message}")
        raise

    except ValueError as e:
        raise ValidationError(f"Invalid value: {e}")

def retry_deserialize(data, max_retries=3):
    """带重试的反序列化"""
    import time

    for attempt in range(max_retries):
        try:
            return safe_deserialize(data)
        except (SerializationException, ValidationError) as e:
            if attempt == max_retries - 1:
                raise
            print(f"Attempt {attempt + 1} failed, retrying...")
            time.sleep(0.1 * (attempt + 1))  # 指数退避

if __name__ == "__main__":
    valid_data = {
        "version": "1.0",
        "name": "Test",
        "values": [1, 2, 3]
    }

    # 序列化和反序列化
    serialized = serialize(valid_data)
    result = retry_deserialize(serialized)
    print(f"Result: {result}")

    # 处理错误
    invalid_data = b"invalid protobuf data"
    try:
        result = deserialize(invalid_data)
    except SerializationException as e:
        print(f"Caught: {e}")

关键洞察：序列化绑定的核心挑战是错误处理。对于格式错误的数据，必须提供清晰的错误信息（行号、列号、具体原因）。FlatBuffers 提供零拷贝反序列化，非常适合性能敏感的场景。Protocol Buffers 有成熟的生态系统，适合跨语言的 IDL 定义场景。