线程和多处理模块之间有什么区别？

问题描述：

我正在学习如何使用Python 中的threading和multiprocessing模块并行运行某些操作并加快我的代码速度。

threading.Thread()我发现很难（可能是因为我没有任何理论背景）理解对象和一之间的区别multiprocessing.Process()。

此外，我不完全清楚如何实例化一个作业队列，并且只有其中 4 个（例如）并行运行，而其他作业则等待资源释放后再执行。

我发现文档中的示例很清楚，但不是很详尽；一旦我尝试使事情稍微复杂化，我就会收到很多奇怪的错误（例如无法腌制的方法等等）。

那么，我应该何时使用threading和multiprocessing模块？

您能否给我提供一些资源来解释这两个模块背后的概念以及如何正确使用它们来完成复杂的任务？

解决方案 1：

Giulio Franco 所说的话对于多线程与多处理而言总体上是正确的。

但是，Python *还有一个问题：有一个全局解释器锁，可防止同一进程中的两个线程同时运行 Python 代码。这意味着，如果您有 8 个核心，并将代码更改为使用 8 个线程，它将无法使用 800% 的 CPU 并以 8 倍的速度运行；它将使用相同的 100% 的 CPU 并以相同的速度运行。（实际上，它会运行得慢一点，因为即使您没有任何共享数据，线程也会产生额外的开销，但现在请忽略这一点。）

但也存在例外。如果您代码中的繁重计算实际上并非发生在 Python 中，而是发生在某些具有自定义 C 代码的库中，这些库可以正确处理 GIL，例如 numpy 应用程序，那么您将从线程中获得预期的性能优势。如果繁重计算由您运行并等待的某个子进程完成，情况也是如此。

更重要的是，有些情况下这并不重要。例如，网络服务器大部分时间都在从网络读取数据包，而 GUI 应用程序大部分时间都在等待用户事件。在网络服务器或 GUI 应用程序中使用线程的一个原因是允许您执行长时间运行的“后台任务”，而无需停止主线程继续为网络数据包或 GUI 事件提供服务。这在 Python 线程中运行得很好。（从技术角度来说，这意味着 Python 线程为您提供并发性，即使它们不提供核心并行性。）

但是如果你用纯 Python 编写 CPU 绑定程序，使用更多线程通常没有帮助。

使用单独的进程不会出现 GIL 带来的上述问题，因为每个进程都有自己独立的 GIL。当然，您仍然需要像使用其他任何语言一样在线程和进程之间进行权衡 — 在进程之间共享数据比在线程之间共享数据更困难、更昂贵，运行大量进程或频繁创建和销毁进程的成本可能很高，等等。但是 GIL 在很大程度上影响了进程之间的平衡，而 C 或 Java 则不是这样。因此，您会发现在 Python 中使用多处理的频率比在 C 或 Java 中高得多。

同时，Python 的“内置电池”理念也带来了一些好消息：只需更改一行代码就可以很容易地编写在线程和进程之间来回切换的代码。

如果您以自包含的“作业”的形式设计代码，这些作业除了输入和输出之外不与其他作业（或主程序）共享任何内容，那么您可以使用该concurrent.futures库围绕线程池编写代码，如下所示：

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    executor.submit(job, argument)
    executor.map(some_function, collection_of_independent_things)
    # ...

您甚至可以获取这些作业的结果并将其传递给进一步的作业，按执行顺序或完成顺序等待事情等等；请阅读有关Future对象的部分了解详情。

现在，如果事实证明您的程序始终使用 100% 的 CPU，而添加更多线程只会使其变慢，那么您就遇到了 GIL 问题，因此您需要切换到进程。您所要做的就是更改第一行：

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:

唯一真正的警告是，您的作业的参数和返回值必须是可 pickle 的（并且 pickle 不会花费太多时间或内存），才能跨进程使用。通常这不是问题，但有时是问题。

但是，如果您的作业不能独立运行怎么办？如果您可以按照将消息从一个作业传递到另一个作业的方式来设计代码，那么这仍然相当容易。您可能必须使用threading.Thread或multiprocessing.Process而不是依赖池。并且您必须明确创建queue.Queue或对象。（还有很多其他选项 — 管道、套接字、带有 flock 的文件……但关键是，如果 Executor 的自动魔力不够，multiprocessing.Queue您必须手动执行某些操作。）

但是，如果您甚至不能依赖消息传递怎么办？如果您需要两个作业都改变相同的结构并查看彼此的更改怎么办？在这种情况下，您将需要进行手动同步（锁定、信号量、条件等），并且如果您想使用进程，则需要显式共享内存对象来启动。这是多线程（或多处理）变得困难的时候。如果您可以避免它，那就太好了；如果你不能，你需要阅读比别人在 SO 答案中投入的更多内容。

从评论中，您想知道 Python 中线程和进程之间的区别。实际上，如果您阅读了 Giulio Franco 的回答和我的回答以及我们所有的链接，应该会涵盖所有内容……但总结肯定是有用的，因此这里是：

1. 线程默认共享数据；进程则不然。

2. 由于 (1) 的原因，在进程之间发送数据通常需要进行 pickle 和 unpickle 操作。**

3. 作为 (1) 的另一个结果，在进程之间直接共享数据通常需要将其放入低级格式，如值、数组和ctypes类型。

4. 流程不受 GIL 约束。

5. 在某些平台（主要是 Windows）上，创建和销毁进程的代价要高得多。

6. 对进程有一些额外的限制，其中一些在不同的平台上有所不同。有关详细信息，请参阅编程指南。

7. 该threading模块不具备该模块的某些功能multiprocessing。（您可以使用该multiprocessing.dummy模块来获取线程之上的大部分缺失 API，或者您可以使用更高级的模块concurrent.futures，而不必担心这一点。）

• 实际上，存在此问题的不是 Python 语言，而是 CPython，即该语言的“标准”实现。其他一些实现没有 GIL，例如 Jython。

** 如果您使用fork启动方法进行多处理（您可以在大多数非 Windows 平台上使用），则每个子进程都会获得父进程在启动子进程时所拥有的任何资源，这也是将数据传递给子进程的另一种方式。

解决方案 2：

单个进程中可以存在多个线程。属于同一进程的线程共享同一内存区域（可以读取和写入完全相同的变量，并且可以相互干扰）。相反，不同的进程位于不同的内存区域中，每个进程都有自己的变量。为了进行通信，进程必须使用其他通道（文件、管道或套接字）。

如果您想要并行化计算，您可能需要多线程，因为您可能希望线程在同一内存上协作。

说到性能，线程的创建和管理速度比进程更快（因为操作系统不需要分配全新的虚拟内存区域），线程间通信通常比进程间通信更快。但线程编程起来更困难。线程可以相互干扰，并且可以写入彼此的内存，但这种情况发生的方式并不总是很明显（由于几个因素，主要是指令重新排序和内存缓存），因此您将需要同步原语来控制对变量的访问。

解决方案 3：

Python 文档引述

我已经在以下位置突出显示了有关进程、线程和 GIL 的关键 Python 文档引述：CPython 中的全局解释器锁 (GIL) 是什么？

进程与线程实验

我做了一些基准测试，以便更具体地显示差异。

在基准测试中，我为8 个超线程CPU 上不同数量的线程计时了 CPU 和 IO 绑定工作。每个线程提供的工作始终相同，因此线程越多意味着提供的总工作量越多。

结果是：

绘制数据。

各部分含义：

• “CPU 限制”：所有线程/进程完成每个线程的固定数量的 CPU 限制工作所需的时间。每个线程的工作量是固定的，因此，例如，当使用 2 个线程时，将完成 2 倍的总工作量。

解释：线程总是比较慢，因为它们在争夺 CPU 锁

• “CPU 限制/线程”：上图除以线程数。这给出了完成每个工作单元所需的平均时间。我们观察到：

+ 对于线程来说，这是常量：2 倍的工作需要 2 倍的时间才能完成。因此，它根本没有并行化。
+ 对于进程，该比率会下降，直到 4 倍，然后保持不变。因此，它可以很好地并行化到 4 倍，并且能够更快地运行，但无法扩展到更高。


由于我使用的是 4 核 8 超线程机器，所以我预计可以扩展到 8 倍。


与达到预期 8 倍加速的 C POSIX CPU 绑定工作进行对比：time(1) 输出中的“real”、“user”和“sys”是什么意思？


TODO：我不知道这个原因，肯定有其他 Python 效率低下的问题在起作用。

• “线程/进程比率”：上面两行的比率。这非常清楚地显示了 4 倍加速限制。

• “IO 绑定”：与“CPU 绑定”相同，但具有 IO 绑定任务

结论：

• 对于 CPU 密集型工作，多处理总是更快，大概是因为 GIL

• 对于 IO 绑定工作。两者的速度完全相同

测试代码：

#!/usr/bin/env python3

import multiprocessing
import threading
import time
import sys

def cpu_func(result, niters):
    '''
    A useless CPU bound function.
    '''
    for i in range(niters):
        result = (result * result * i + 2 * result * i * i + 3) % 10000000
    return result

class CpuThread(threading.Thread):
    def __init__(self, niters):
        super().__init__()
        self.niters = niters
        self.result = 1
    def run(self):
        self.result = cpu_func(self.result, self.niters)

class CpuProcess(multiprocessing.Process):
    def __init__(self, niters):
        super().__init__()
        self.niters = niters
        self.result = 1
    def run(self):
        self.result = cpu_func(self.result, self.niters)

class IoThread(threading.Thread):
    def __init__(self, sleep):
        super().__init__()
        self.sleep = sleep
        self.result = self.sleep
    def run(self):
        time.sleep(self.sleep)

class IoProcess(multiprocessing.Process):
    def __init__(self, sleep):
        super().__init__()
        self.sleep = sleep
        self.result = self.sleep
    def run(self):
        time.sleep(self.sleep)

if __name__ == '__main__':
    cpu_n_iters = int(sys.argv[1])
    sleep = 1
    cpu_count = multiprocessing.cpu_count()
    input_params = [
        (CpuThread, cpu_n_iters),
        (CpuProcess, cpu_n_iters),
        (IoThread, sleep),
        (IoProcess, sleep),
    ]
    header = ['nthreads']
    for thread_class, _ in input_params:
        header.append(thread_class.__name__)
    print(' '.join(header))
    for nthreads in range(1, 2 * cpu_count):
        results = [nthreads]
        for thread_class, work_size in input_params:
            start_time = time.time()
            threads = []
            for i in range(nthreads):
                thread = thread_class(work_size)
                threads.append(thread)
                thread.start()
            for i, thread in enumerate(threads):
                thread.join()
            results.append(time.time() - start_time)
        print(' '.join('{:.6e}'.format(result) for result in results))

GitHub 上游 + 在同一目录上绘制代码。

在 Ubuntu 18.10、Python 3.6.7 上进行测试，使用联想 ThinkPad P51 笔记本电脑，CPU：Intel Core i7-7820HQ CPU（4 核/8 线程），RAM：2x Samsung M471A2K43BB1-CRC（2x 16GiB），SSD：Samsung MZVLB512HAJQ-000L7（3,000 MB/s）。

可视化在给定时间内正在运行的线程

这篇文章https://rohanvarma.me/GIL/告诉我，每当一个线程被调度时，你都可以运行一个回调，并且target=参数threading.Thread也是一样multiprocessing.Process。

这样我们就可以准确查看每次运行的线程。完成后，我们会看到类似这样的内容（我制作了这个特定的图表）：

            +--------------------------------------+
            + Active threads / processes           +
+-----------+--------------------------------------+
|Thread   1 |********     ************             |
|         2 |        *****            *************|
+-----------+--------------------------------------+
|Process  1 |***  ************** ******  ****      |
|         2 |** **** ****** ** ********* **********|
+-----------+--------------------------------------+
            + Time -->                             +
            +--------------------------------------+

这表明：

• 线程完全由 GIL 序列化

• 进程可以并行运行

解决方案 4：

我相信这个链接以一种优雅的方式回答了您的问题。

简而言之，如果您的一个子问题必须等待另一个子问题完成，则多线程是不错的选择（例如，在 I/O 密集型操作中）；相反，如果您的子问题确实可能同时发生，则建议使用多处理。但是，您不会创建比核心数量更多的进程。

解决方案 5：

以下是 Python 2.6.x 的一些性能数据，这些数据让人质疑在 IO 受限的情况下线程比多处理更高效这一观点。这些结果来自 40 处理器的 IBM System x3650 M4 BD。

IO 绑定处理：进程池比线程池表现更好

>>> do_work(50, 300, 'thread','fileio')
do_work function took 455.752 ms

>>> do_work(50, 300, 'process','fileio')
do_work function took 319.279 ms

CPU 密集型处理：进程池比线程池表现更好

>>> do_work(50, 2000, 'thread','square')
do_work function took 338.309 ms

>>> do_work(50, 2000, 'process','square')
do_work function took 287.488 ms

这些并不是严格的测试，但它们告诉我，与线程相比，多处理的性能并非完全不佳。

上述测试中交互式 Python 控制台使用的代码

from multiprocessing import Pool
from multiprocessing.pool import ThreadPool
import time
import sys
import os
from glob import glob

text_for_test = str(range(1,100000))

def fileio(i):
 try :
  os.remove(glob('./test/test-*'))
 except : 
  pass
 f=open('./test/test-'+str(i),'a')
 f.write(text_for_test)
 f.close()
 f=open('./test/test-'+str(i),'r')
 text = f.read()
 f.close()


def square(i):
 return i*i

def timing(f):
 def wrap(*args):
  time1 = time.time()
  ret = f(*args)
  time2 = time.time()
  print '%s function took %0.3f ms' % (f.func_name, (time2-time1)*1000.0)
  return ret
 return wrap

result = None

@timing
def do_work(process_count, items, process_type, method) :
 pool = None
 if process_type == 'process' :
  pool = Pool(processes=process_count)
 else :
  pool = ThreadPool(processes=process_count)
 if method == 'square' : 
  multiple_results = [pool.apply_async(square,(a,)) for a in range(1,items)]
  result = [res.get()  for res in multiple_results]
 else :
  multiple_results = [pool.apply_async(fileio,(a,)) for a in range(1,items)]
  result = [res.get()  for res in multiple_results]


do_work(50, 300, 'thread','fileio')
do_work(50, 300, 'process','fileio')

do_work(50, 2000, 'thread','square')
do_work(50, 2000, 'process','square')

解决方案 6：

好吧，Giulio Franco 回答了大部分问题。我将进一步阐述消费者-生产者问题，我想这将为您使用多线程应用程序的解决方案指明正确的方向。

fill_count = Semaphore(0) # items produced
empty_count = Semaphore(BUFFER_SIZE) # remaining space
buffer = Buffer()

def producer(fill_count, empty_count, buffer):
    while True:
        item = produceItem()
        empty_count.down();
        buffer.push(item)
        fill_count.up()

def consumer(fill_count, empty_count, buffer):
    while True:
        fill_count.down()
        item = buffer.pop()
        empty_count.up()
        consume_item(item)

您可以从以下位置阅读有关同步原语的更多信息：

 http://linux.die.net/man/7/sem_overview
 http://docs.python.org/2/library/threading.html

伪代码如上。我想你应该搜索生产者-消费者问题来获得更多参考。