问题描述：

我有一个foo消耗大量内存的函数，我想并行运行它的多个实例。

假设我有一个具有 4 个物理核心的 CPU，每个核心有两个逻辑核心。

我的系统有足够的内存来容纳 4 个并行实例foo，但没有足够的内存来容纳 8 个。此外，由于这 8 个核心中有 4 个是逻辑核心，因此我也不期望使用所有 8 个核心会比仅使用 4 个物理核心带来太多收益。

所以我只想foo在 4 个物理核心上运行。换句话说，我想确保执行multiprocessing.Pool(4)（由于内存限制，4 是我可以在此机器上容纳的最大并发运行次数）将作业分派到四个物理核心（而不是，例如，两个物理核心及其两个逻辑子代的组合）。

如何在 python 中做到这一点？

编辑：

我之前使用过一个代码示例，multiprocessing但我对库不了解，所以为了避免混淆，我删除了它。

解决方案 1：

我知道这个话题现在已经很老了，但是在谷歌中输入“多处理逻辑核心”时它仍然作为第一个答案出现......我觉得我必须给出一个额外的答案，因为我可以看到 2018 年（甚至更晚）的人们可能会很容易在这里感到困惑（有些答案确实有点令人困惑）

我认为没有比这里更好的地方来警告读者注意上述一些答案，所以很抱歉让这个话题再次陷入争论。

--> 要计算 CPU 数量（逻辑/物理），请使用 PSUTIL 模块

例如，对于 4 个物理核心 / 8 线程 i7，它将返回

import psutil 
psutil.cpu_count(logical = False)

4

psutil.cpu_count(logical = True)

8

就这么简单。

这样你就不用担心操作系统、平台、硬件本身或其他什么了。我确信它比 multiprocessing.cpu_count() 好得多，后者有时会给出奇怪的结果，至少从我自己的经验来看是这样。

--> 要使用 N 个物理核心（由您选择），请使用 YUGI 描述的多处理模块

只需计算您有多少个物理进程，启动一个由 4 个工作进程组成的多处理池。

或者您也可以尝试使用 joblib.Parallel() 函数

2018 年的 joblib 不是 python 标准发行版的一部分，而只是 Yugi 描述的多处理模块的一个包装器。

--> 大多数情况下，不要使用超过可用数量的核心（除非你对一个非常具体的代码进行了基准测试，并证明了它是值得的）

错误信息比比皆是：“如果您指定的核心数量超过可用核心数量，操作系统就会处理问题”。这绝对是 100% 错误的。如果您使用的核心数量超过可用核心数量，您将面临巨大的性能下降。例外情况是工作进程受 IO 限制。因为操作系统调度程序会尽力以同样的注意力处理每项任务，定期从一个任务切换到另一个任务，并且根据操作系统的不同，它可能会花费高达 100% 的工作时间在进程之间切换，这将是灾难性的。

不要只相信我：尝试一下，对它进行基准测试，你会看到它有多清晰。

是否可以决定代码是在逻辑核心还是物理核心上执行？

如果您问这个问题，这意味着您不了解物理和逻辑核心的设计方式，所以也许您应该多了解一下处理器的架构。

例如，如果您想在核心 3 而不是核心 1 上运行，那么我猜确实有一些解决方案，但只有当您知道如何编写操作系统的内核和调度程序时才可用，而我认为如果您问这个问题，情况并非如此。

如果您在 4 个物理/8 个逻辑处理器上启动 4 个 CPU 密集型进程，则调度程序会将每个进程归因于 1 个不同的物理核心（并且 4 个逻辑核心将保持未使用或使用较少）。但是在 4 个逻辑/8 个线程处理器上，如果处理单元为 (0,1) (1,2) (2,3) (4,5) (5,6) (6,7)，那么进程在 0 或 1 上执行没有区别：它们是同一个处理单元。

据我所知（但专家可以确认，也许它与非常具体的硬件规格也不同），我认为在 0 或 1 上执行代码没有或几乎没有区别。在处理单元 (0,1) 中，我不确定 0 是逻辑的而 1 是物理的，反之亦然。根据我的理解（可能是错误的），两者都是来自同一处理单元的处理器，它们只是共享缓存/对硬件（包括 RAM）的访问，并且 0 并不比 1 更像物理单元。

除此之外，您还应该让操作系统来决定。因为操作系统调度程序可以利用某些平台（例如 i7、i5、i3...）上存在的硬件逻辑核心加速功能，而其他一些您无权控制的功能可能对您真正有帮助。

如果您在 4 个物理核心/8 个逻辑核心上启动 5 个 CPU 密集型任务，则行为将变得混乱，几乎不可预测，主要取决于您的硬件和操作系统。调度程序将尽最大努力。几乎每次，您都会面临非常糟糕的性能。

让我们暂时假设我们仍在谈论 4（8）经典架构：因为调度程序会尽力而为（因此经常切换属性），具体取决于您正在执行的进程，在 5 个逻辑核心上启动可能比在 8 个逻辑核心上启动更糟糕（至少他知道无论如何一切都会 100% 使用，所以无论怎样他都不会试图避免它，不会切换得太频繁，因此不会因为切换而浪费太多时间）。

然而，99% 可以肯定（但请在您的硬件上进行基准测试以确保），如果您使用的物理核心多于可用核心，几乎任何多处理程序的运行速度都会变慢。

很多事情都会影响性能...程序、硬件、操作系统的状态、它使用的调度程序、你今天早上吃的水果、你妹妹的名字...如果你对某件事有疑问，就对其进行基准测试，没有其他简单的方法可以查看你的性能是否下降。有时信息学真的很奇怪。

--> 大多数情况下，额外的逻辑核心在 PYTHON 中确实是无用的（但并非总是如此）

在 Python 中，有两种主要方法可以完成真正的并行任务。

• 多处理（无法利用逻辑核心）

• 多线程（可以利用逻辑核心）

例如并行运行 4 个任务

--> multiprocessing 将创建 4 个不同的 Python 解释器。对于每个解释器，您都必须启动一个 Python 解释器、定义读/写权限、定义环境、分配大量内存等。让我​​们这样说：您将从 0 开始启动一个全新的程序实例。这可能需要大量时间，因此您必须确保这个新程序能够运行足够长的时间，这样才值得。

如果您的程序有足够的工作（比如说，至少几秒钟的工作），那么由于操作系统将占用 CPU 的进程分配到不同的物理核心上，因此它可以正常工作，并且您可以获得很多性能，这很棒。而且由于操作系统几乎总是允许进程之间进行通信（尽管速度很慢），它们甚至可以交换（一点点）数据。

--> 多线程则不同。在您的 Python 解释器中，它只会创建少量内存，许多 CPU 可以共享这些内存，并同时对其进行处理。它的生成速度要快得多（在旧计算机上生成新进程有时需要几秒钟，而生成线程则只需极短的时间）。您不会创建新进程，而是创建更轻量的“线程”。

线程可以在线程之间非常快速地共享内存，因为它们实际上是在同一块内存上一起工作的（而当与不同的进程一起工作时，必须复制/交换）。

但是：为什么我们不能在大多数情况下使用多线程？它看起来很方便？

Python 有一个非常大的限制：Python 解释器中一次只能执行一行 Python 代码，这被称为 GIL（全局解释器锁）。因此，大多数情况下，使用多线程甚至会降低性能，因为不同的线程必须等待才能访问同一资源。对于纯计算处理（没有 IO），如果您的代码是纯 Python，多线程是无用的，甚至更糟。但是，如果您的线程涉及任何等待 IO，多线程可能会非常有益。

--> 为什么使用多处理时不应该使用逻辑核心？

逻辑核心没有自己的内存访问。它们只能在内存访问和托管物理处理器的缓存上工作。例如，同一处理单元的逻辑核心和物理核心很可能（并且确实经常使用）同时在缓存内存的不同位置上使用相同的 C/C++ 函数。这确实使处理速度大大加快。

但是……这些是 C/C++ 函数！Python 是一个大型 C/C++ 包装器，需要的内存和 CPU 比其等效的 C++ 代码多得多。在 2018 年，无论您想做什么，两个大型 Python 进程都需要比单个物理+逻辑单元所能承受的内存和缓存读写多得多的内存和缓存读写，也比等效的 C/C++ 真正多线程代码所消耗的多得多。这几乎总是会导致性能下降。请记住，处理器缓存中不可用的每个变量都需要 1000 倍的时间才能读取内存。如果您的缓存已经完全被 1 个 Python 进程占满，猜猜如果您强制 2 个进程使用它会发生什么：它们会一次使用一个，并永久切换，导致每次切换时数据都会被愚蠢地刷新和重新读取。当从内存读取或写入数据时，您可能会认为您的 CPU“正在”工作，但事实并非如此。它正在等待数据！什么也不做。

-->那么您如何利用逻辑核心呢？

就像我说的，由于全局解释器锁的存在，默认 Python 中没有真正的多线程（因此没有真正使用逻辑核心）。您可以在程序的某些部分强制删除 GIL，但我认为，如果您不清楚自己在做什么，明智的做法是不要触碰它。

删除 GIL 无疑已成为大量研究的主题（参见尝试这样做的实验性 PyPy 或 Cython 项目）。

目前，还没有真正的解决方案，因为这个问题比看起来要复杂得多。

我承认，还有另一种可行的解决方案：

• 使用 C 语言编写函数

• 使用 ctype 将其包装在 python 中

• 使用 python 多线程模块调用包装的 C 函数

这将 100% 地发挥作用，并且您将能够使用 Python 中的所有逻辑核心，并实现多线程和真实功能。GIL 不会打扰您，因为您不会执行真正的 Python 函数，而是执行 C 函数。

例如，一些像 Numpy 这样的库可以在所有可用的线程上工作，因为它们是用 C 编写的。但是如果你到了这一点，我一直认为直接用 C/C++ 编写你的程序是明智的，因为这与原始的 Python 精神相去甚远。

--> 不要总是使用所有可用的物理核心

我经常看到有人说“好吧，我有 8 个物理核心，所以我将使用 8 个核心来完成我的工作”。这通常有效，但有时却是一个糟糕的想法，特别是如果你的工作需要大量 I/O 时。

尝试使用 N-1 个核心（再次强调，特别是对于 I/O 要求高的任务），您将发现，在 100% 的时间里，按每个任务/平均而言，单个任务在 N-1 个核心上总是运行得更快。事实上，您的计算机会制造很多不同的东西：USB、鼠标、键盘、网络、硬盘等……即使在工作站上，也会在后台随时执行您不知道的定期任务。如果您不让 1 个物理核心来管理这些任务，您的计算将被定期中断（从内存中清除/重新放入内存），这也会导致性能问题。

你可能会想“好吧，后台任务只会使用 5% 的 CPU 时间，所以还剩下 95%”。但事实并非如此。

处理器每次只处理一项任务。每次切换时，都会浪费大量时间将所有内容放回内存缓存/注册表中。然后，如果出于某种奇怪的原因，操作系统调度程序过于频繁地进行这种切换（这是您无法控制的），所有这些计算时间都将永远丢失，您对此无能为力。

如果（有时会发生）由于某些未知原因，此调度程序问题影响了 30 个任务的性能，而不是 1 个任务的性能，则可能会导致非常有趣的情况，即在 29/30 个物理核心上工作的速度可能比在 30/30 个物理核心上工作的速度要快得多

CPU 越多并不总是最好的

使用 multiprocessing.Pool 时，经常会使用在进程之间共享的 multiprocessing.Queue 或管理器队列，以允许它们之间进行一些基本通信。有时（我肯定说了 100 次，但我还是要重复一遍），以硬件相关的方式，可能会发生（但您应该针对特定应用程序、代码实现和硬件对其进行基准测试）使用更多 CPU 可能会在使进程进行通信/同步时造成瓶颈。在这些特定情况下，在较低的 CPU 数量上运行可能会很有趣，甚至可以尝试将同步任务转移到更快的处理器上（这里我当然是在谈论在集群上运行的科学密集型计算）。由于多处理通常用于集群，因此您必须注意到，为了节省能源，集群的频率通常会降低。因此，单核性能可能非常糟糕（通过大量 CPU 来平衡），当你将代码从本地计算机（核心少，单核性能高）扩展到集群（核心多，单核性能较低）时，问题会变得更加严重，因为你的代码瓶颈取决于 single_core_perf/nb_cpu 比率，这有时真的很烦人

每个人都想使用尽可能多的 CPU。但在这种情况下，基准测试是强制性的。

典型情况（例如在数据科学中）是让 N 个进程并行运行，并且您希望将结果汇总到一个文件中。由于您不能等待作业完成，因此您可以通过特定的写入器进程来完成。写入器将在其 multiprocessing.Queue（单核和硬盘受限进程）中推送的所有内容写入输出文件中。N 个进程填满了 multiprocessing.Queue。

很容易想象，如果你有 31 个 CPU 将信息写入一个非常慢的 CPU，那么你的性能就会下降（如果你克服了系统处理临时数据的能力，可能会崩溃）

--> 带回家的信息

• 使用 psutil 来计算逻辑/物理处理器，而不是 multiprocessing.cpu_count() 或任何其他方法

• 多处理只能在物理核心上工作（或者至少对其进行基准测试以证明在您的情况下它不正确）

• 多线程将在逻辑核心上工作，但你必须用 C 语言编写并包装你的函数，或者删除全局锁解释器（每次你这样做，世界上某个地方就会有一只小猫惨死）

• 如果你尝试在纯 Python 代码上运行多线程，性能会大幅下降，因此 99% 的时间你都应该使用多处理

• 除非你的进程/线程有可以利用的长时间暂停，否则永远不要使用超过可用核心的数量，如果你想尝试，请进行适当的基准测试

• 如果您的任务是 I/O 密集型的，您应该让 1 个物理核心来处理 I/O，如果您有足够的物理核心，这将是值得的。对于多处理实现，需要使用 N-1 个物理核心。对于经典的双向多线程，这意味着使用 N-2 个逻辑核心。

• 如果你需要更多的性能，可以尝试 PyPy（尚未准备好投入生产）或 Cython，甚至可以用 C 语言编写代码

最后但并非最不重要且最重要的一点：如果您真的追求性能，您绝对应该始终进行基准测试，而不是猜测任何事情。基准测试通常会揭示您不知道的奇怪的平台/硬件/驱动程序非常具体的行为。

解决方案 2：

注意：此方法在 Windows 上不起作用，并且仅在 Linux 上进行了测试。

使用multiprocessing.Process：

使用时，为每个进程分配一个物理核心非常容易Process()。您可以创建一个 for 循环，迭代每个核心，并使用将新进程分配给新核心taskset -p [mask] [pid]：

import multiprocessing
import os

def foo():
    return

if __name__ == "__main__" :
    for process_idx in range(multiprocessing.cpu_count()):
        p = multiprocessing.Process(target=foo)
        os.system("taskset -p -c %d %d" % (process_idx % multiprocessing.cpu_count(), os.getpid()))
        p.start()

我的工作站上有 32 个核心，因此我将部分结果放在这里：

pid 520811's current affinity list: 0-31
pid 520811's new affinity list: 0
pid 520811's current affinity list: 0
pid 520811's new affinity list: 1
pid 520811's current affinity list: 1
pid 520811's new affinity list: 2
pid 520811's current affinity list: 2
pid 520811's new affinity list: 3
pid 520811's current affinity list: 3
pid 520811's new affinity list: 4
pid 520811's current affinity list: 4
pid 520811's new affinity list: 5
...

如您所见，这里是每个进程的先前和新的亲和力。第一个进程适用于所有核心 (0-31)，然后分配给核心 0，第二个进程默认分配给核心 0，然后其亲和力更改为下一个核心 (1)，依此类推。

使用multiprocessing.Pool：

警告：此方法需要调整pool.py模块，因为据我所知，您无法从中提取 pid 。此外，此更改已在和上Pool()进行了测试。python 2.7`multiprocessing.__version__ = '0.70a1'`

在 中Pool.py，找到调用该方法的行_task_handler_start()。在下一行中，您可以使用以下命令将池中的进程分配给每个“物理”核心（我将 放在import os这里，以便读者不会忘记导入它）：

import os
for worker in range(len(self._pool)):
    p = self._pool[worker]
    os.system("taskset -p -c %d %d" % (worker % cpu_count(), p.pid))

你就大功告成了。测试：

import multiprocessing

def foo(i):
    return

if __name__ == "__main__" :
    pool = multiprocessing.Pool(multiprocessing.cpu_count())
    pool.map(foo,'iterable here')

结果：

pid 524730's current affinity list: 0-31
pid 524730's new affinity list: 0
pid 524731's current affinity list: 0-31
pid 524731's new affinity list: 1
pid 524732's current affinity list: 0-31
pid 524732's new affinity list: 2
pid 524733's current affinity list: 0-31
pid 524733's new affinity list: 3
pid 524734's current affinity list: 0-31
pid 524734's new affinity list: 4
pid 524735's current affinity list: 0-31
pid 524735's new affinity list: 5
...

请注意，此修改将pool.py作业循环分配给核心。因此，如果您分配的作业多于 CPU 核心，则最终会在同一个核心上拥有多个作业。

编辑：

OP 正在寻找的是能够pool()启动特定核心上的池。为此multiprocessing需要进行更多调整（首先撤消上述更改）。

警告：

不要尝试复制粘贴函数定义和函数调用。仅复制粘贴应该添加的部分self._worker_handler.start()（您将在下面看到）。请注意，我的multiprocessing.__version__版本是'0.70a1'，但只要您添加需要添加的内容，这并不重要：

multiprocessing的pool.py：

在定义中添加一个cores_idx = None参数__init__()。在我的版本中，添加后如下所示：

def __init__(self, processes=None, initializer=None, initargs=(),
             maxtasksperchild=None,cores_idx=None)

您还应该在后面添加以下代码self._worker_handler.start()：

if not cores_idx is None:
    import os
    for worker in range(len(self._pool)):
        p = self._pool[worker]
        os.system("taskset -p -c %d %d" % (cores_idx[worker % (len(cores_idx))], p.pid))

multiprocessing的__init__.py：

cores_idx=None在 in 的定义Pool()以及Pool()返回部分的其他函数调用中添加一个参数。在我的版本中它看起来像：

def Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None,cores_idx=None):
    '''
    Returns a process pool object
    '''
    from multiprocessing.pool import Pool
    return Pool(processes, initializer, initargs, maxtasksperchild,cores_idx)

您已完成。以下示例仅在核心 0 和 2 上运行 5 个工作线程池：

import multiprocessing


def foo(i):
    return

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=5,cores_idx=[0,2])
    pool.map(foo,'iterable here')

结果：

pid 705235's current affinity list: 0-31
pid 705235's new affinity list: 0
pid 705236's current affinity list: 0-31
pid 705236's new affinity list: 2
pid 705237's current affinity list: 0-31
pid 705237's new affinity list: 0
pid 705238's current affinity list: 0-31
pid 705238's new affinity list: 2
pid 705239's current affinity list: 0-31
pid 705239's new affinity list: 0

当然，multiprocessing.Poll()通过删除该cores_idx参数，您仍然能够获得其通常的功能。

解决方案 3：

我找到了一个不需要更改 Python 模块源代码的解决方案。它使用了此处建议的方法。运行该脚本后，可以通过执行以下操作来检查是否只有物理核心处于活动状态：

lscpu

在 bash 中返回：

CPU(s):                8
On-line CPU(s) list:   0,2,4,6
Off-line CPU(s) list:  1,3,5,7
Thread(s) per core:    1

[可以从python内部运行上面链接的脚本]。无论如何，运行上述脚本后，在 python 中输入以下命令：

import multiprocessing
multiprocessing.cpu_count()

返回 4。