将具有无限线程的多线程代码转换为具有最大同时运行线程数的多线程代码

Question

我有一个通过多线程执行某个功能的脚本。现在，让并行运行的线程数量与拥有 CPU 核心的数量一样多是很有趣的。 现在使用 threading.thread 语句的当前代码 (1:) 创建了 1000 个线程并同时运行它们。 我想把它变成同时只运行固定数量的线程（例如 8 个）的东西，并将其余的放入队列中，直到正在执行的线程/cpu 核心可以免费使用。

1：

import threading

nSim = 1000

def simulation(i):
    print(str(threading.current_thread().getName()) + ': '+ str(i))

if __name__ == '__main__':
    threads = [threading.Thread(target=simulation,args=(i,)) for i in range(nSim)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

Q1：代码2：是否符合我的描述？ （同时运行最大线程数的多线程）是否正确？ （我想是的，但我不是 100% 确定）

Q2：现在代码同时启动1000个线程，在8个线程上执行。有没有办法只在执行线程/cpu 内核空闲时才启动一个新线程（这样我就没有 990 个线程调用从一开始就等待执行？

Q3：有没有办法跟踪哪个cpu-core执行了哪个线程？只是为了证明代码正在做它应该做的事情。

2：

import threading
import multiprocessing

print(multiprocessing.cpu_count())
from concurrent.futures import ThreadPoolExecutor

nSim = 1000

def simulation(i):
    print(str(threading.current_thread().getName()) + ': '+ str(i))

if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=8) as executor:
        for i in range (nSim):
            res = executor.submit(simulation, i)
            print(res.result())

Answer 1

A1：为了限制可以同时访问某些资源的线程数，你可以使用 threading.Semaphore 实际上 1000 个线程不会给你带来巨大的速度提升，每个进程的推荐线程数是 mp.cpu_count() *1 或某些文章中的 mp.cpu_count()*2。另请注意，线程适用于 python 中的 IO 操作，但由于 GIL，不适用于计算。

A2。如果您只想同时运行 8 个线程，为什么需要这么多线程？仅创建 8 个线程，然后在任务准备就绪时为它们提供任务，为此您需要使用线程安全的 queue.Queue()。但是在您的具体示例中，您可以执行以下操作，以便在模拟功能中使用 while 每个线程运行 250 次测试，顺便说一下，在这种情况下您不需要 Semaphore。

A3。当我们谈论多线程时，您有一个具有多个线程的进程。

导入线程
进口时间
将多处理导入为 mp

def 模拟（i，_s）：
# s 是线程。Semaphore()
与_s：
打印(str(threading.current_thread().getName()) + ': ' + str(i))
time.sleep(3)

如果 name == 'ma​​in':
print("核心数：{}".format(mp.cpu_count()))
# 在一些文章中推荐线程数为 mp.cpu_count()*1 或 mp.cpu_count()*2
nSim = 25

s = threading.Semaphore(4) # max number of threads which can work simultaneously with resource is 4 threads = [threading.Thread(target=simulation, args=(i, s, )) for i in range(nSim)] for t in threads: t.start() # just to prove that all threads are active in the start and then their number decreases when the work is done for i in range(6): print("Active threads number {}".format(threading.active_count())) time.sleep(3)

Answer 2

A1：不，您的代码提交了一个任务，在res 中接收到一个Future，然后调用result，它等待 得到结果。只有在前一个任务完成后，才会给线程一个新任务。一次只有一个工作线程真正在工作。

查看ThreadPool.map（实际上是Pool.map）而不是submit，以便在工作人员之间分配任务。

A2：这里最多只使用 8 个线程（worker 的数量）。如果使用map，可能会存储 1000 个任务的输入数据（需要内存），但不会创建额外的线程。

A3：我不知道。线程不绑定到内核，它可以在它们之间快速切换。