将 for 循环代码转换为具有最大线程数的多线程代码

Question

背景：我正在尝试使用 python-dymola 界面进行 100 次 dymola 模拟。我设法在 for 循环中运行它们。现在我希望它们在多线程时运行，这样我就可以并行运行多个模型（这会快得多）。由于可能没有人使用该界面，因此我编写了一些简单的代码，也显示了我的问题：

1：将 for 循环转换为运行到另一个 for 循环中的定义，但 def 和 for 循环共享相同的变量 'i'。

2：将for循环变成定义，使用多线程来执行。 for 循环一个一个地运行命令。我想同时使用最多 x 个线程并行运行它们。结果应该和执行for循环时一样

示例代码：

import os

nSim = 100
ndig='{:01d}'

for i in range(nSim):
    os.makedirs(str(ndig.format(i)))

请注意，创建的目录的名称只是 for 循环中的数字（这很重要）。现在，我不想使用 for 循环，而是喜欢使用多线程创建目录（注意：对于这个简短的代码可能不感兴趣，但是当调用和执行 100 个模拟模型时，使用多线程肯定很有趣。线程）。

所以我从我认为的一些简单的事情开始，将 for 循环转换为一个函数，然后在另一个 for 循环中运行，并希望得到与上面的 for 循环代码相同的结果，但得到了这个错误： AttributeError: 'NoneType' 对象没有属性 'start' （注：我是从这个开始的，因为我之前没有使用def-statement，而且线程包也是新的。之后我会向多线程发展。）

1：

import os

nSim = 100
ndig='{:01d}'

def simulation(i):
    os.makedirs(str(ndig.format(i)))

for i in range(nSim):
    simulation(i=i).start

在那次失败之后，我尝试进化到多线程（将 for 循环转换为具有相同功能但使用多线程的东西，并通过并行运行代码而不是一个一个地运行代码，并且最大数量线程）：

2：

import os
import threading


nSim = 100
ndig='{:01d}'

def simulation(i):
    os.makedirs(str(ndig.format(i)))

if __name__ == '__main__':
    i in range(nSim)
    simulation_thread[i] = threading.Thread(target=simulation(i=i))
    simulation_thread[i].daemon = True
    simulation_thread[i].start()

不幸的是，该尝试也失败了，现在我得到了错误： NameError: name 'i' is not defined

有人对问题 1 或 2 有建议吗？

Answer 1

这两个例子都不完整。这是一个完整的例子。请注意，target 传递了函数的名称 target=simulation 及其参数的元组 args=(i,)。不要调用函数target=simulation(i=i)，因为这只是传递函数的结果，在这种情况下相当于target=None。

import threading

nSim = 100

def simulation(i):
    print(f'{threading.current_thread().name}: {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()

输出：

Thread-1: 0
Thread-2: 1
Thread-3: 2
 .
 .
Thread-98: 97
Thread-99: 98
Thread-100: 99

请注意，您通常不需要比 CPU 更多的线程，您可以从 multiprocessing.cpu_count() 获得这些线程。您可以使用创建线程池并使用queue.Queue 发布线程执行的工作。 Python Queue documentation 中有一个示例。

Answer 2

1. 不能像这样调用.start

   simulation(i=i).start
   

   在非线程对象上。此外，您还必须导入模块

2. 您似乎忘记在循环中添加“for”并缩进代码

   i in range(nSim)
   simulation_thread[i] = threading.Thread(target=simulation(i=i))
   simulation_thread[i].daemon = True
   simulation_thread[i].start()
   

   到

   for i in range(nSim):
       simulation_thread[i] = threading.Thread(target=simulation(i=i))
       simulation_thread[i].daemon = True
       simulation_thread[i].start()
   

Answer 3

如果您想在池中拥有最大线程数，并运行队列中的所有项目。我们可以继续@mark-tolonen 回答并这样做：

import threading
import queue
import time

def main():
    size_of_threads_pool = 10
    num_of_tasks = 30
    task_seconds = 1
    
    q = queue.Queue()

    def worker():
        while True:
            item = q.get()
            print(my_st)
            print(f'{threading.current_thread().name}: Working on {item}')
            time.sleep(task_seconds)
            print(f'Finished {item}')
            q.task_done()

    my_st = "MY string"
    threads = [threading.Thread(target=worker, daemon=True) for i in range(size_of_threads_pool)]
    for t in threads:
        t.start()


    # send the tasks requests to the worker
    for item in range(num_of_tasks):
        q.put(item)

    # block until all tasks are done
    q.join()
    print('All work completed')

    # NO need this, as threads are while True, so never will stop..
    # for t in threads:
    #    t.join()


if __name__ == '__main__':
    main()

这将使用 10 个线程运行 30 个任务，每个任务 1 秒。 所以总时间是 3 秒。

$ time python3 q_test.py 
...
All work completed

real  0m3.064s
user  0m0.033s
sys   0m0.016s

编辑：我找到了另一个用于异步执行可调用对象的高级接口。

使用concurrent.futures，参见文档中的example：

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

注意max_workers=5，它将告诉最大线程数，并且 请注意您可以使用的 for 循环 for url in URLS。