Ch 15: 项目迁移策略

本章目标

• 制定合理的迁移计划
• 使用性能分析工具定位瓶颈
• 实现 Python 与 C++ 混合编程
• 正确选择和使用数据结构映射
• 验证迁移结果

15.1 迁移策略概述

从 Python 迁移到 C++ 是一个系统工程，需要合理的规划和执行：

15.1.1 为什么需要迁移

Python 项目迁移到 C++ 的典型原因：

原因	说明	适用场景
性能瓶颈	CPU 密集型操作成为瓶颈	数值计算、图像处理、游戏引擎
延迟要求	毫秒级延迟不可接受	高频交易、实时系统、游戏
内存控制	需要精确控制内存布局	嵌入式、驱动程序、游戏
部署依赖	无法部署 Python 运行时	桌面应用、移动端、游戏
并发性能	GIL 限制多线程性能	服务器处理、科学计算

15.1.2 迁移策略选择

Python 项目
    │
    ├─► 增量迁移（推荐）
    │      │
    │      ├─► 识别性能关键代码
    │      ├─► 用 C++ 重写核心模块
    │      ├─► 通过 pybind11 暴露接口
    │      └─► Python 调用 C++，逐步替换
    │
    └─► 完全重写
           │
           ├─► 适合小型项目
           ├─► 风险高，工作量大
           └─► 但可以获得最佳架构

15.1.3 迁移检查清单

迁移前评估：

• 识别性能瓶颈（用 profiler）
• 评估 Python 标准库替代方案
• 确定 C++ 目标标准（C++17/20/23）
• 选择构建系统（CMake、Bazel）
• 制定测试策略

迁移执行：

• 设置开发环境（C++ 编译器、IDE）
• 配置持续集成
• 创建 C++ 基础架构（构建、测试）
• 逐模块迁移
• 保持 Python 接口兼容

迁移后验证：

• 功能测试（回归测试）
• 性能基准测试
• 内存泄漏检查
• 代码覆盖率

15.2 识别性能瓶颈

15.2.1 Python 性能分析工具

import cProfile
import pstats
import io

# 1. cProfile - 函数级分析
def profile_function():
    profiler = cProfile.Profile()
    profiler.enable()

    # 你的代码
    result = some_heavy_computation()

    profiler.disable()

    # 输出统计
    s = io.StringIO()
    stats = pstats.Stats(profiler, stream=s)
    stats.sort_stats('cumulative')  # 按累计时间排序
    stats.print_stats(20)  # 前 20 行
    print(s.getvalue())

# 2. line_profiler - 行级分析
# pip install line_profiler
from line_profiler import LineProfiler

def profile_lines():
    profiler = LineProfiler()
    profiler.add_function(heavy_function)
    profiler.run('profile_function()')
    profiler.print_stats()

# 3. memory_profiler - 内存分析
# pip install memory_profiler
from memory_profiler import profile

@profile
def memory_intensive():
    data = [i ** 2 for i in range(1000000)]
    return data

15.2.2 C++ 性能分析工具

#include <chrono>
#include <iostream>
#include <fstream>
#include <iomanip>

// 1. Chrono 简单计时
void simple_timing() {
    auto start = std::chrono::steady_clock::now();

    // 代码
    heavy_computation();

    auto end = std::chrono::steady_clock::now();
    auto duration = std::chrono::duration<double, std::milli>(end - start);
    std::cout << "Elapsed: " << duration.count() << " ms" << std::endl;
}

// 2. 高精度测量
class ScopedTimer {
public:
    ScopedTimer(const std::string& name) : name_(name), start_(std::chrono::high_resolution_clock::now()) {}
    ~ScopedTimer() {
        auto end = std::chrono::high_resolution_clock::now();
        auto duration = std::chrono::duration<double, std::nano>(end - start_);
        std::cout << name_ << ": " << duration.count() / 1e6 << " ms" << std::endl;
    }
private:
    std::string name_;
    std::chrono::high_resolution_clock::time_point start_;
};

void with_scoped_timer() {
    ScopedTimer timer("Total operation");
    // 代码
}

// 3. 性能基准框架
// Google Benchmark (https://github.com/google/benchmark)
#include <benchmark/benchmark.h>

static void BM_StringCreation(benchmark::State& state) {
    for (auto _ : state) {
        std::string s("hello");
        benchmark::DoNotOptimize(s);
    }
}
BENCHMARK(BM_StringCreation);

BENCHMARK_MAIN();

15.2.3 常见性能瓶颈模式

Python 常见瓶颈：

# 1. 循环中的 Python 虚拟姬调用
result = []
for i in range(1000000):  # Python 循环慢
    result.append(process(i))

# C++ 替代：向量化或重写为 C++
# NumPy 向量化
import numpy as np
result = np.arange(1000000)
result = process_vectorized(result)

# 2. 字符串拼接
s = ""
for i in range(10000):
    s += str(i)  # 每次创建新字符串

# 替代：join 或 list 再 join
parts = [str(i) for i in range(10000)]
s = "".join(parts)

# 3. 全局解释器锁（GIL）
import threading
def cpu_bound():
    while True:
        x = 1 + 2

threads = [threading.Thread(target=cpu_bound) for _ in range(4)]
for t in threads:
    t.start()
# 不能真正并行！

C++ 优化策略：

// 1. 使用reserve避免重新分配
std::vector<int> v;
v.reserve(1000000);  // 预先分配
for (int i = 0; i < 1000000; ++i) {
    v.push_back(i);
}

// 2. 移动语义
std::vector<int> source = get_data();
// 拷贝改为移动
std::vector<int> dest = std::move(source);

// 3. 避免不必要的拷贝
void process(const std::vector<int>& data) {  // const ref，不拷贝
    // 只读访问
}

// 4. 使用emplace而非push_back
std::vector<std::string> v;
v.emplace_back("hello");  // 直接构造
v.push_back("hello");     // 拷贝或移动

15.3 混合编程

Python 与 C++ 混合编程是增量迁移的关键技术：

15.3.1 pybind11 入门

pybind11 是最流行的 Python-C++ 绑定库：

cpp_module.cpp

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <vector>
#include <string>

namespace py = pybind11;

// 简单函数
int add(int a, int b) {
    return a + b;
}

// 使用标准容器
std::vector<int> filter_even(const std::vector<int>& nums) {
    std::vector<int> result;
    for (int n : nums) {
        if (n % 2 == 0) result.push_back(n);
    }
    return result;
}

// Python 类对应 C++ 类
class Calculator {
public:
    Calculator(int initial) : value_(initial) {}

    void add(int v) { value_ += v; }
    void multiply(int v) { value_ *= v; }
    int value() const { return value_; }

private:
    int value_;
};

PYBIND11_MODULE(cpp_module, m) {
    m.doc() = "Example pybind11 module";

    // 导出函数
    m.def("add", &add, "Add two numbers");
    m.def("filter_even", &filter_even, "Filter even numbers");

    // 导出类
    py::class_<Calculator>(m, "Calculator")
        .def(py::init<int>(), "Initialize with value")
        .def("add", &Calculator::add, "Add to value")
        .def("multiply", &Calculator::multiply, "Multiply value")
        .def("value", &Calculator::value, "Get current value");
}

15.3.2 编译 pybind11 模块

# 方法 1: 使用 setup.py
# setup.py
from setuptools import setup, Extension

cpp_module = Extension(
    'cpp_module',
    sources=['cpp_module.cpp'],
    include_dirs=['/path/to/pybind11/include'],
    extra_compile_args=['-std=c++17'],
)

setup(
    name='cpp_module',
    ext_modules=[cpp_module],
)

setup.py

from setuptools import setup, Extension
from pybind11 import get_include
import sys

ext_modules = [
    Extension(
        'cpp_module',
        ['cpp_module.cpp'],
        include_dirs=[get_include()],
        language='c++',
        extra_compile_args=['-std=c++17'],
    ),
]

setup(
    name='cpp_module',
    ext_modules=ext_modules,
)

# 编译安装
python setup.py build_ext --inplace

# 或使用 pip
pip install -e .

15.3.3 CMake 构建 pybind11

CMakeLists.txt

cmake_minimum_required(VERSION 3.15)
project(cpp_module)

set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

find_package(pybind11 REQUIRED)
find_package(Python3 REQUIRED COMPONENTS Interpreter Development)

pybind11_add_module(cpp_module cpp_module.cpp)

# 安装到 site-packages
install(TARGETS cpp_module LIBRARY DESTINATION ${Python3_SITELETTINGS})

15.3.4 Python 调用 C++ 模块

import cpp_module

# 调用函数
result = cpp_module.add(1, 2)
print(f"1 + 2 = {result}")  # 3

# 使用标准容器
numbers = list(range(10))
evens = cpp_module.filter_even(numbers)
print(f"Even numbers: {evens}")  # [0, 2, 4, 6, 8]

# 使用类
calc = cpp_module.Calculator(10)
calc.add(5)
calc.multiply(2)
print(f"Result: {calc.value()}")  # 30

15.3.5 高级 pybind11 特性

#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <pybind11/functional.h>
#include <vector>
#include <string>

namespace py = pybind11;

// 返回 None
void process_with_optional(int* input, int* output) {
    // output 是 nullptr 表示返回 None
}

// 处理 Python 异常
void risky_function() {
    throw std::runtime_error("Something went wrong");
}

PYBIND11_MODULE(cpp_module, m) {
    // 关键字参数
    m.def("func", [](int a, int b) { return a + b; },
          py::arg("a"), py::arg("b"));

    // 默认参数
    m.def("func_with_default", [](int x, int y = 10) { return x + y; },
          py::arg("x") = 5, py::arg("y") = 10);

    // 文档字符串
    m.def("documented", [](int x) { return x * 2; },
          "Doubles the input value");

    // 抛出异常
    m.def("check_positive", [](int x) {
        if (x <= 0) {
            throw py::value_error("Value must be positive");
        }
        return x;
    });

    // lambda
    m.def("square", [](int x) { return x * x; });
}

15.4 数据结构迁移对照

15.4.1 容器选择指南

Python 容器	C++ 容器	何时使用
list	std::vector	默认选择，需要索引访问
list（两端操作多）	std::deque	两端频繁添加删除
tuple	std::tuple	异构固定大小
dict	std::unordered_map	键值对，快速查找
dict（有序）	std::map	需要有序遍历
set	std::unordered_set	唯一性，快速查找
set（有序）	std::set	需要有序遍历
list（堆）	std::priority_queue	优先级队列

15.4.2 详细对照

// Python list → C++ vector
std::vector<int> vec = {1, 2, 3};
vec.push_back(4);
int first = vec[0];
int last = vec.back();
vec.pop_back();

// Python dict → C++ unordered_map
std::unordered_map<std::string, int> dict = {{"a", 1}, {"b", 2}};
dict["c"] = 3;
int val = dict["a"];
auto it = dict.find("b");
if (it != dict.end()) {
    val = it->second;
}

// Python set → C++ unordered_set
std::unordered_set<int> set = {1, 2, 3};
set.insert(4);
bool exists = set.count(2) > 0;

// Python tuple → C++ tuple
std::tuple<int, std::string, double> t = {1, "hello", 3.14};
int a = std::get<0>(t);
std::string b = std::get<1>(t);

// 结构化绑定（C++17）
auto [id, name, score] = t;

15.4.3 选择决策树

需要键值映射？
├─ 是 → 需要有序遍历？
│       ├─ 是 → std::map
│       └─ 否 → std::unordered_map
└─ 否 → 需要唯一性？
        ├─ 是 → 需要有序遍历？
        │       ├─ 是 → std::set
        │       └─ 否 → std::unordered_set
        └─ 否 → 主要操作是？
                ├─ 尾部添加/索引访问 → std::vector
                ├─ 两端添加删除 → std::deque
                └─ 中间插入删除 → std::list

15.5 迁移案例：数值计算模块

15.5.1 Python 原版

compute.py

import numpy as np

def matrix_multiply(a: list, b: list) -> list:
    """矩阵乘法"""
    n = len(a)
    result = [[0.0] * n for _ in range(n)]
    for i in range(n):
        for j in range(n):
            for k in range(n):
                result[i][j] += a[i][k] * b[k][j]
    return result

def find_primes(n: int) -> list:
    """找素数"""
    if n < 2:
        return []
    is_prime = [True] * (n + 1)
    is_prime[0] = is_prime[1] = False
    for i in range(2, int(n**0.5) + 1):
        if is_prime[i]:
            for j in range(i*i, n+1, i):
                is_prime[j] = False
    return [i for i in range(2, n+1) if is_prime[i]]

# 性能瓶颈测试
import time
start = time.time()
result = matrix_multiply([[1,2],[3,4]], [[5,6],[7,8]])
print(f"Matrix: {time.time() - start:.4f}s")

15.5.2 C++ 优化版

compute.cpp

#include <pybind11/pybind11.h>
#include <vector>
#include <numeric>
#include <cmath>

namespace py = pybind11;

std::vector<std::vector<double>> matrix_multiply(
    const std::vector<std::vector<double>>& a,
    const std::vector<std::vector<double>>& b) {
    size_t n = a.size();
    std::vector<std::vector<double>> result(n, std::vector<double>(n, 0.0));

    for (size_t i = 0; i < n; ++i) {
        for (size_t k = 0; k < n; ++k) {
            for (size_t j = 0; j < n; ++j) {
                result[i][j] += a[i][k] * b[k][j];
            }
        }
    }
    return result;
}

std::vector<int> find_primes(int n) {
    if (n < 2) return {};

    std::vector<bool> is_prime(n + 1, true);
    is_prime[0] = is_prime[1] = false;

    int limit = static_cast<int>(std::sqrt(n));
    for (int i = 2; i <= limit; ++i) {
        if (is_prime[i]) {
            for (int j = i * i; j <= n; j += i) {
                is_prime[j] = false;
            }
        }
    }

    std::vector<int> primes;
    for (int i = 2; i <= n; ++i) {
        if (is_prime[i]) primes.push_back(i);
    }
    return primes;
}

PYBIND11_MODULE(compute, m) {
    m.def("matrix_multiply", &matrix_multiply, "Matrix multiplication");
    m.def("find_primes", &find_primes, "Find all primes up to n");
}

15.5.3 Python 使用 C++ 模块

use_compute.py

import compute
import time

# 使用 C++ 版本
start = time.time()
result = compute.matrix_multiply([[1,2],[3,4]], [[5,6],[7,8]])
print(f"C++ Matrix: {time.time() - start:.6f}s")
print(f"Result: {result}")

start = time.time()
primes = compute.find_primes(100000)
print(f"C++ Primes: {time.time() - start:.4f}s")
print(f"Found {len(primes)} primes")

15.6 验证与测试

15.6.1 测试策略

test_compute.cpp

#include <catch2/catch.hpp>
#include "compute.cpp"

TEST_CASE("Matrix multiplication") {
    std::vector<std::vector<double>> a = {{1, 2}, {3, 4}};
    std::vector<std::vector<double>> b = {{5, 6}, {7, 8}};

    auto result = matrix_multiply(a, b);

    REQUIRE(result[0][0] == Approx(19));
    REQUIRE(result[0][1] == Approx(22));
    REQUIRE(result[1][0] == Approx(43));
    REQUIRE(result[1][1] == Approx(50));
}

TEST_CASE("Find primes") {
    auto primes = find_primes(10);

    REQUIRE(primes.size() == 4);
    REQUIRE(primes == std::vector<int>{2, 3, 5, 7});
}

15.6.2 Python 回归测试

test_python_wrapper.py

import cpp_module
import compute as py_compute

def test_matrix_multiply():
    a = [[1.0, 2.0], [3.0, 4.0]]
    b = [[5.0, 6.0], [7.0, 8.0]]

    cpp_result = cpp_module.matrix_multiply(a, b)
    py_result = py_compute.matrix_multiply(a, b)

    assert cpp_result == py_result

def test_find_primes():
    for n in [10, 100, 1000, 10000]:
        cpp_result = set(cpp_module.find_primes(n))
        py_result = set(py_compute.find_primes(n))
        assert cpp_result == py_result

章节总结

• 增量迁移是降低风险的最佳策略
• 用 cProfile 和 line_profiler 定位 Python 瓶颈
• 用 Chrono 和 benchmark 进行 C++ 性能测量
• pybind11 是 Python-C++ 混合编程的首选工具
• 数据结构选择对性能影响巨大
• 保持 Python 接口兼容，实现平滑过渡
• 充分的测试覆盖确保迁移正确性

---

下章预告：ch16 最佳实践。