如何让Python运行速度像C++一样快

最近一个月我研究了一下Python的C API，也尝试写了一个扩展，然后惊奇的发现这玩意儿大大提高了Python的运行速度，甚至一度超过了C++的运行速度。对比图如下：

这里的实验环境是这样的，操作系统是Windows 10，CPU是从core i7（64位8核），编译器是MSC 14.35.32215.0，即Visual Studio 2022的开发环境。Python的API是3.8+。

两边都是同样的逻辑，先生成2048个数，然后分别让每个数加一，再打印这些数。Python的用时是0.4秒，而C++的用时是0.8秒，Python的速度显然已经超过了C++的速度！那么是如何做到的呢？

本篇文章先通过介绍一个项目streamcpy，再引入Python C API的概念，最后解答文章标题的问题——如何让Python运行速度像C++一样快

项目介绍

这个项目叫streamcpy，是一个Python版的Stream API (流式计算API)。项目地址点 GitHub - littlebutt/streamcpy: The Stream API in Python. 。灵感来自于Java8 的 Stream API ，它可以在处理大量数据时使用短路运算（shortcircuit computation）或者延迟计算（lazy computation），并且不保留中间运算结果，从而实现运算提速。

安装方法

先引下载streamcpy的包，并在项目中导入。如果是Windows用户，streamcpy包可以直接通过pip下载并安装，比如：

python -m pip install streamcpy

如果是Linux/Mac用户则需要编译安装（Windows当然也可以）。编译安装需要有python的dev环境以及C语言编译器。

说明：如果是Windows用户，在安装Python的时候会自动下载相应版本的dev环境，即header文件和dll文件，而Linux/Mac用户，则需要安装python-dev这个软件包。编译的时候将头文件和动态链接库引入环境即可。 C语言编译器没有明确的要求，但是建议使用相应平台的编译器，比如Windows用户使用MSVC（即Visual Studio集成的），Linux用户用gcc，Mac用户用clang（即XCode集成的）。

首先，按照惯例克隆项目：

git clone https://github.com/littlebutt/streamcpy.git

然后在根目录中执行如下指令编译项目：

python setup.py build

最后安装项目：

python setup.py install

在python环境中输入以下内容：

import streamcpy
streamcpy.__version__

>>> 1.0

如果不报错则说明OK了。

使用方法

简单介绍一下它的使用方法，如果熟悉Java8的Stream API的话可以跳过下面这部分内容。不过，考虑到Python的语言特点，有部分API做了优化。

of方法

类似于Java，可以用 Iterable类型 的对象构造一个Stream。这个Iterable就是需要处理的数据 data ，比如这是一个list对象：

streamcpy.Stream.of(['foo', 'bar'])
                .for_each(lambda item: print(item))
>>> foo
>>> bar

也可以是tuple对象，比如：

streamcpy.Stream.of(('foo', 'bar'))
                .for_each(lambda item: print(item))
>>> foo
>>> bar

甚至还可以是generator对象和file对象，比如：

def gen():
    data = ['foo', 'bar']
    for item in data:
        yield item

streamcpy.Stream.of(gen())
                .for_each(lambda item: print(item))
>>> foo
>>> bar

streamcpy.Stream.of(open('foo.txt', 'r'))
                .for_each(lambda item: print(item))

for_each方法

这个方法需要传入一个Callable参数，并将数据 data 中的每一个元素作为这个Callable的参数执行。比如上面例子中，for_each方法就是将数据 data 都打印出来。当然也可以有其他用法，比如：

result = []

streamcpy.Stream.of(['foo', 'bar'])
                .for_each(lambda item: result.append(item))

这个方法是一个终止操作（Terminal Operation）。

这个概念来自于Java，即有的方法（操作）只能被用作调用链的最后一个方法（操作），而有的方法（操作）只能被用作调用链的中间方法（操作）。

filter方法

这个方法用来过滤数据 data 。同样，它也需要传入一个Callable对象，而且这个Callable对象需要返回True或False来表示是否保留，比如：

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .filter(lambda item: item > 2)
                .for_each(lambda item: print(item))
>>> 3
>>> 4
>>> 5

这个方法是一个中间操作（Intermidiate Operation）。

map方法

这个方法用来修改数据 data 。同样，它也需要传入一个Callable对象，而且这个Callable对象需要有返回值，比如：

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .map(lambda item: item * item)
                .for_each(lambda item: print(item))
>>> 1
>>> 4
>>> 9
>>> 16
>>> 25

这个方法是一个中间操作。

collect方法

这个方法用来将数据 data 收集到列表list内，它需要传入一个list对象，比如：

result = []

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .collect(result)
result

>>> [1, 2, 3, 4, 5]

注意，这个collect方法是通过追加的方式添加数据的，也就是说如果result里面有数据，那么这些数据仍然会被保留，不会被覆盖。

这个方法是一个终止操作。

distinct方法

这个方法用来去除数据 data 中重复的值，它不需要任何参数，比如：

streamcpy.Stream.of([1, 2, 3, 3, 3, 4])
                .distinct()
                .for_each(lambda item: print(item))
>>> 1
>>> 2
>>> 3
>>> 4

注意，这个方法是通过对象的 __hash__来判断是否相同的，如果两个对象尽管内容不一样，但事实哈希值相同则仍视为相同。

这个方法是一个中间操作。

limit方法

这个方法用来获取数据 data 的前N个数据，它需要一个 int 类型的参数N，比如：

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .limit(3)
                .for_each(lambda item: print(item))
>>> 1
>>> 2
>>> 3

注意，如果传入的参数不是int（及其子类）则会抛出TypeError异常。

这个方法是一个中间操作。

reduce方法

这个方法用来对数据 data 执行 reduce 操作。简单来说，就是将 data 内的前两个元素作为REDUCE方法的参数，并将计算结果和后续元素作为新的两个参数再次作为REDUCE方法的参数进行计算，直到所有元素被消费并返回结果。它需要一个Callable对象作为参数REDUCE，且这个REDUCE需要两个对象作为参数。比如：

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .reduce(lambda x, y: x + y)
>>> 15                                             # 15 = 1 + 2 + 3 + 4 + 5

注意，如果输入的参数REDUCE不符合要求则会抛出相应的异常，如果给定的数据 data 含有0个元素，则抛出 RuntimeError 异常，如果给定的数据 data 含有1个元素，则将这个元素作为返回结果。

这个方法是一个终止操作。

sorted方法

这个方法用来将数据 data 根据给定的SORTED方法排序，它需要一个Callable对象作为参数SORTED，且这个REDUCE需要两个对象作为参数以及只能返回int类型（及其子类）。这个返回值用来表示大小关系，若大于0则说明前者大，若小于0则说明后者大，否则说明一样大。比如：

streamcpy.Stream.of([3, 4, 1, 2, 5])
                .sorted(lambda x, y: x - y)
                .for_each(lambda item: print(item))
>>> 1
>>> 2
>>> 3
>>> 4
>>> 5

注意，如果输入的参数SORTED不符合要求则会抛出相应的异常，如果给定的数据 data 含有0个元素，则抛出 RuntimeError 异常，如果给定的数据 data 含有1个元素，则将这个元素作为返回结果。

这个方法是一个中间操作。

max/min方法

这个方法用来返回数据 data 的最大值/最小值，它也需要一个Callable对象作为参数，这个Callable是用来获取被比较的对象，这个对象必须实现 __cmp__。比如：

streamcpy.Stream.of([3, 4, 1, 2, 5])
                .max(lambda x:x)
>>> 5

streamcpy.Stream.of([{'name': 'foo', 'id': 2}, {'name': 'bar', 'id': 3}, {'name': 'foobar', 'id': 1}])
                .max(lambda x: x.id)
>>> {'name': 'bar', 'id': 3}

streamcpy.Stream.of([3, 4, 1, 2, 5])
                .min(lambda x:x)
>>> 1

注意，如果输入的参数不符合要求则会抛出相应的异常，如果给定的数据 data 含有0个元素，则抛出 RuntimeError 异常，如果给定的数据 data 含有1个元素，则将这个元素作为返回结果。

这个方法是一个终止操作。

count方法

这个方法返回数据 data 的长度，它不需要任何参数，比如：

streamcpy.Stream.of([3, 4, 1, 2, 5]).count()

>>> 5

这个方法是一个终止操作。

any_match/all_match方法

这个方法返回数据 data 是否对存在/任意一个元素都满足给定条件CONDITION，它接受一个Callable参数CONDITION，且这个CONDITION必须返回True或者False表示是否满足条件。比如：

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .any_match(lambda item: item > 4)
>>> True

streamcpy.Stream.of([1, 2, 3, 4, 5])
                .all_match(lambda item: item > 4)
>>> False

这个方法是一个终止操作。

以上就是streamcpy的全部用法，所有用法都经过测试，如果存在BUG也欢迎在Github页面提交issue。

实现过程

现在重点来了！这么一个扩展是怎么实现的呢？它为什么比C++快？

首先回答第一个问题。这个包虽然是Python包的形式，但它完全是由C语言写的。下面通过streamcpy的例子简单介绍一下CPython的C API 。

CPython是Python解释器的一种实现，是用C写的。我们平常所说的Python一般默认是指CPython。除了CPython之外，还有Jython（用Java写的）、Pypi（用Python写的）和IronPython（用C#写的）等。

C API本质上是对Python语句的重写，也就是所有用Python写的内容都可以用其对应版本的C API再写一遍达到原封不动的效果，但是运行速度会大大提高。除此以外，C API的版本还需要用户手动处理一下内存管理。

现在拿streamcpy举例。在这个包中，除了有Stream这个对象还有一个Pipeline对象，这个对象是以链表的形式组织的，其结构如下：

class Pipeline:
    op_type: str
    op_method: Any
    next: "Pipeline"
    
    @staticmethod
    def append(pl: Pipeline, op_type: str, op_method: Any) -> None
        ...
    
    @staticmethod
    def execute(pl: Pipeline, data: Iterable[Any]) -> Any
        ...

这里为了方便理解，对Python代码做了type hint，相关说明见这里。

这个类的逻辑是这样的，op_type用于存放操作的种类，比如map、filter或者for_each等，用str表示。而op_method用于存放操作的内容，一般是一个Callable对象，比如说函数、生成器等，但也有特殊情况，比如当op_type为collect时则为List。next则为指向下一个Pipeline的指针。所以整个构造形式如下:

这个类有两个静态函数，append是将给定Pipeline链表的末尾新增一个节点，它接受三个参数，分别是给定的Pipeline的头节点，操作类型op_type以及操作内容op_method。execute是将给定的data根据Pipeline执行一遍，这一步才是真正的执行。这里的逻辑也是参考Java8的Stream API实现的。当然后者更复杂。

更多源码可以参考项目repostreamcpy/python at main · littlebutt/streamcpy · GitHub

现在，我们再看一下C语言的版本，注意两者的区别：

static PyTypeObject Pipeline_type;

typedef struct _Pipeline {
    PyObject_HEAD

    int op_type;      // op_type

    PyObject* op_method;    // op_method

    struct _Pipeline* next; // next

} Pipeline;

static void
Pipeline_dealloc(Pipeline* pl)
{
    ...
}

static int
Pipeline_append(Pipeline* pl, const int op_type, PyObject* op_method)
{
    ...
}

int
_Pipeline_execute_map(PyObject** data, PyObject* op_method)
{
    ...
}

static PyMemberDef Pipeline_members[] = {
    {"op_type", T_INT, offsetof(Pipeline, op_type), 0, PyDoc_STR("a value representing the method type")},
    {"op_method", T_OBJECT, offsetof(Pipeline, op_method), 0, PyDoc_STR("a callable value for eval the given data")},
    {"next", T_OBJECT, offsetof(Pipeline, next), 0, PyDoc_STR("a pointer pointing to the next Pipeline")},
    {NULL}
};

static PyTypeObject Pipeline_type = {
    PyVarObject_HEAD_INIT(NULL, 0)
    "streampy.Pipeline",                    /* tp_name */
    ...
    (destructor) Pipeline_dealloc,          /* tp_dealloc */
    ...
    Pipeline_members,                       /* tp_members */
    ...
};

PyMODINIT_FUNC PyInit_streamcpy() {
    ...
}

同样我也是节选了一部分代码，可以看出它由两部分组成，一部分是PyObject本身（这里是Pipeline），它表示一个Python对象（动态初始化），另一个是PyTypeObject，它表示一个Python的类型（静态初始化）。

这里需要指明PyTypeObject也是PyObject的子类，并且也可以动态初始化。我们熟知的Python自带的包datetime、functool等都是动态初始化它定义的对象的。

PyObject就像我们定义的Python类一样，可以给他定义成员和方法，比如在Pipeline这个struct内，我同样定义了一个int类型的op_type表示操作类型，一个PyObject*类型的op_method表示操作内容，而next表示指向下一个Pipeline的指针。

PyObject*表示一个Python的对象，经过特定的方法初始化以后，它可以暴露给最终用户使用，比如在streamcpy中，用户可以直接创建一个Pipeline的Python对象，虽然我不建议这样做。

除了Pipeline的成员外，我还定义了两个外部方法Pipeline_append和Pipeline_execute分别对应Python版本的append和execute，这里不再赘述。但与之不同的是，我还定义了一个Pipeline_dealloc方法，它是Pipeline的析构函数。与Java不同，Python的垃圾回收机制（GC）是通过引用计数（reference count）实现的，被引用的对象可以通过Py_INCREF让其引用计数加一，解除引用后的对象可以通过Py_DECREF让其引用计数减一，当引用计数为零的时候，Python虚拟机就会调用这里的析构函数释放内存。

最后，我初始化了一个PyTypeObject，这个对象（struct）是用来刻画这个Pipeline的。比如这里我给它绑定了类名（"streampy.Pipeline"）、成员变量（Pipeline_members）和析构函数（Pipeline_dealloc）。当然我还省略了其他插槽（slot）的数据，比如__repr__函数、__hash__函数等，它们都是在这里被定义的。

最后，再通过C API的创建模块函数PyInit_streamcpy将这个模块创建出来。由于篇幅有限，这里就不继续介绍Python的C API了，如果有机会我也会单独开一个专栏讲解如何用C编写Python扩展实现加速 。

反思总结

事情到这里还远没有结束，因为我们还没有回答文章标题中的问题。但是我们可以先回答这个问题——

为什么C语言写的Python程序比Python写的快？

原因是用户使用Python的C API可以优化Python的 编译速度 和 部分运行速度 。具体来说，它可以让Python代码跳过编译阶段（Python代码->Python字节码），然后在运行阶段（Python字节码->二进制机器码）可以跳过字节码解释。

我们先看一下CPython的执行流。如图：

普通的.py文件运行必须要经过以上各个阶段，而C语言写的程序不仅可以跳过编译阶段（因为不涉及Python语法解析），甚至还可以直接到中间字节码转译（因为CPython的字节码本身就是在调用C API，即C语言的实现）。少了花费时长最长的几个步骤运行速度（Runtime）自然就提高了。

或许和很多人印象中的Python不一样，很多人以为Python是纯脚本语言（类似Javascript），不存在编译和执行两个阶段。其实这个论断不准确，具体要看Python的运行方式。我们平时编写的.py文件的方式是存在编译和执行两个阶段的。读者可以通过运行python -m py_compile <target>查看中间字节码。关于CPython源码解析我也不会在这里讲，或许有机会新开专栏讲解。

最后，为什么C语言写的Python程序就是比C++快？

这个问题涉及到C++实现了，文章开头给定的程序是通过MSVC编译运行的。由于使用了STL，MSVC对应的STL实现在这里。这里的vector在push_back的过程中涉及到扩容，而且通过cout打印的过程中执行效率也比较低下（数据流的形式），尽管是微软给出的STL方案。换句话说，是Python背后的C语言强大，能够使程序运行速度超过C++！

当然，在和朋友交流过程中也发现了例外，如果用GCC（GNU C）编译并运行的话C++版本的代码运行更快，甚至超过几个量级。

文章标题的问题就在这里告一段落了，继续深究下去我水平也不够。如果前面有描述不当的地方敬请斧正。原创不易，看到这里就请各位点个赞吧。