Python 队列 - 最多有 n 个线程在运行

Question

场景:

我有一个非常大的数据库模型迁移正在进行中，用于新构建，我正在研究如何将当前实时数据从 Web 应用程序迁移到本地测试数据库的样板。

我想在 python 中设置一个脚本，该脚本将同时处理我的模型迁移。我的模型实例有 from_legacy 和 to_legacy 方法。到目前为止，我加载了所有实例并为每个实例创建了 threads，每个线程都从核心 threading 模块中继承了 run 方法只是进行转换并保存结果。

我想让程序中的主循环构建这些线程实例的一大堆，然后开始一个一个地处理它们，在它工作时最多同时运行 10 个，并提供下一个在其他人完成迁移时进行处理。

我想不通的是如何正确利用队列来做到这一点？如果每个线程代表迁移的完整任务，我是否应该先加载所有实例，然后创建一个 Queue 并将 maxsize 设置为 10，并且只跟踪当前正在运行的队列？也许是这样的？

currently_running = Queue()
for model in models:
  task = Migrate(models) #this is subclassed thread
  currently_running.put(task)
  task.start()

在这种情况下，依靠 put 调用来阻止它的容量？如果我要走这条路，我将如何调用 task_done？

或者更确切地说，队列是否应该包含所有任务(而不仅仅是开始的任务)并使用 join 来阻止完成？在线程队列上调用 join 是否会启动包含的线程？

解决“最多有 N 个正在运行的线程”问题的最佳方法是什么？队列应该扮演什么角色？

Answer 1

Although not documented , multiprocessing 模块有一个 ThreadPool 类，顾名思义，它创建了一个线程池。它共享 same API as the multiprocessing.Pool class .

然后您可以使用 pool.apply_async 将任务发送到线程池:

import multiprocessing.pool as mpool

def worker(task):
    # work on task
    print(task)     # substitute your migration code here.

# create a pool of 10 threads
pool = mpool.ThreadPool(10)
N = 100

for task in range(N):
    pool.apply_async(worker, args = (task, ))

pool.close()
pool.join()