Python多线程与多进程并行的量化应用解析

在Python编程中，我们还未深入探讨过如何优化模拟计算的执行性能。虽然NumPy、SciPy和pandas在处理向量化代码时极为高效，但在构建事件驱动系统时，这些工具往往难以充分发挥作用。那么是否存在其他加速代码运行的方法？答案是肯定的——但存在一些关键限制！

本文将系统解析Python程序中可引入的并行计算模型。这些模型尤其适用于无需共享状态的模拟场景，例如期权定价所需的蒙特卡洛模拟，以及算法交易中多参数的回测模拟。

我们将重点探讨Threading（线程）库与Multiprocessing（进程）库的技术实现与应用边界。

Python并发编程的底层限制

初学者在尝试使用多线程优化CPU密集型代码时，最常提出的疑问是："为何使用多线程后程序运行反而变慢？"

理论上，多线程代码应能利用多核处理器的额外核心提升性能。然而，由于CPython解释器的全局解释器锁机制，真正的多线程并行在Python中受到根本性限制。

GIL的存在源于Python解释器的非线程安全特性。该机制对所有线程访问Python对象时实施全局强制锁定，确保任一时刻仅有一个线程能获取Python对象或C API的锁。解释器每执行约100条Python字节码或遇到可能阻塞的I/O操作时，会重新获取此锁。因此，对于CPU密集型任务，Threading库无法带来性能提升，而Multiprocessing库则能显著提高执行效率。

并行库实现方案

接下来我们将通过一个示例，对比两种并行库的实际优化效果。

Threading库的适用场景

虽然CPython解释器不支持通过多线程实现真正的多核并行，但Python仍提供Threading库。若无法利用多核，该库的价值何在？

许多涉及网络编程或数据输入/输出的程序常受网络或I/O限制。这意味着解释器需要等待处理来自网络或硬盘等"远程"数据源的操作结果，这类访问速度远低于本地内存或CPU缓存读取。

因此，当需要访问多个数据源时，为每个待访问数据项创建独立线程可有效提升效率。例如网络爬虫场景：单个线程爬取多个URL时，下载延迟远超CPU处理能力，形成严重I/O瓶颈。通过为每个下载任务分配独立线程，程序可并行获取多个数据源，并在下载完成后统一整合结果，从而消除顺序等待延迟。此时程序性能主要受限于客户端/服务器的带宽。

然而，金融计算往往属于CPU密集型任务，涉及大量数值运算（如大规模数值线性代数求解或蒙特卡洛模拟中的随机统计抽样）。对此类任务，Python的Threading库受GIL限制无法提供性能增益。

实战代码解析

以下通过"列表追加数值"的示例演示多线程实现。每个线程创建独立列表并添加随机数，该设计专为展示CPU密集型场景。

实现逻辑分为三步：

1.导入threading模块

2.创建list_append函数，接收三个参数：列表大小（count）、任务标识id（这在我们需要向控制台写入调试信息时会很有用）、目标列表out_list（是我们要向其中追加随机数的列表）

3.__main__函数创建了一个大小为10^7的列表，并使用两个threads线程来执行工作。然后，它创建了一个名为 jobs的列表，用于存储各个独立的线程。threading.Thread 对象将list_append 函数作为一个参数传入，然后将其追加到jobs列表中。

# thread_test.pyimport randomimport threadingdef list_append(count, id, out_list):    """    Creates an empty list and then appends a     random number to the list 'count' number    of times. A CPU-heavy operation!    """    for i in range(count):        out_list.append(random.random())if __name__ == "__main__":    size = 10000000   # Number of random numbers to add    threads = 2   # Number of threads to create    # Create a list of jobs and then iterate through    # the number of threads appending each thread to    # the job list     jobs = []    for i in range(0, threads):        out_list = list()        thread = threading.Thread(target=list_append(size, i, out_list))        jobs.append(thread)    # Start the threads (i.e. calculate the random number lists)    for j in jobs:        j.start()    # Ensure all of the threads have finished    for j in jobs:        j.join()    print "List processing complete."

time python thread_test.py

List processing complete.real    0m2.003suser    0m1.838ssys     0m0.161s

需要注意的是，用户态时间user与系统态时间sys之和约等于实际运行时间real，这表明使用 Threading 库并未带来性能提升。若真有效，实际耗时本应显著减少。在并发编程中，这两类时间通常分别被称为 CPU 时间 与 挂钟时间（实际耗时）。

多进程库：突破GIL的并行方案

要真正利用几乎所有现代消费级处理器中的多核优势，我们可以转而采用Multiprocessing（多进程）库。尽管其语法与Threading库极为相似，但底层机制存在根本差异。

Multiprocessing库会为每个并行任务创建独立的操作系统进程。这一设计巧妙地绕过了GIL的限制：每个进程拥有自己独立的Python解释器及对应的GIL。因此，各个进程可被分配至不同的处理器核心执行，并在所有进程完成后重新整合结果。

然而，这种方案也存在一些局限。创建额外进程会引入I/O开销，因为数据需要在不同处理器间交换传输，这可能增加整体运行时间。但若数据能够严格限定在各进程内部处理，则仍可能获得显著的加速效果。当然，我们始终需要谨记阿姆达尔定律的约束——并行加速受限于程序中必须串行执行的部分。

多进程方案的具体实现

转换为多进程实现仅需两处关键调整：一是修改导入库语句，二是调整multiprocessing.Process的调用方式——此处目标函数的参数需单独传递。除此之外，整体代码结构与前述多线程实现几乎完全相同：

# multiproc_test.pyimport randomimport multiprocessingdef list_append(count, id, out_list):    """    Creates an empty list and then appends a     random number to the list 'count' number    of times. A CPU-heavy operation!    """    for i in range(count):        out_list.append(random.random())if __name__ == "__main__":    size = 10000000   # Number of random numbers to add    procs = 2   # Number of processes to create    # Create a list of jobs and then iterate through    # the number of processes appending each process to    # the job list     jobs = []    for i in range(0, procs):        out_list = list()        process = multiprocessing.Process(target=list_append,                                           args=(size, i, out_list))        jobs.append(process)    # Start the processes (i.e. calculate the random number lists)          for j in jobs:        j.start()    # Ensure all of the processes have finished    for j in jobs:        j.join()    print "List processing complete."

我们可以再次使用类似的控制台命令来计时这段代码的运行：

time python multiproc_test.py

List processing complete.real    0m1.045suser    0m1.824ssys     0m0.231s

从输出数据可以看出：虽然user用户态时间和sys系统态时间基本保持不变，但real实际运行时间却缩短了接近一半。这完全符合预期，因为我们使用了两个进程并行执行。

若将进程数扩展至四个，同时将列表规模减半以便对比（前提是计算机至少拥有四个核心），则会得到如下输出：

List processing complete.real    0m0.540suser    0m1.792ssys     0m0.269s

使用四个进程带来了约 3.8 倍 的加速效果。但我们必须注意，这一结果不宜直接推广到更庞大、更复杂的实际程序中。数据传输开销、硬件缓存层级及其他系统限制，几乎必然会在真实代码中削弱此类性能收益。

在后续的文章中，我们将对事件驱动的回测系统进行改造，引入并行计算技术，从而显著提升其执行多维度参数优化研究的能力。