具有有限 CPU/端口的 Python 多线程处理

Question

我有一个文件夹名称字典，我想并行处理。在每个文件夹下，有一个我想在 series 中处理的文件名数组：

folder_file_dict = {
         folder_name : {
                         file_names_key : [file_names_array]
                       }
        }

最终，我将创建一个名为 folder_name 的文件夹，其中包含名称为 len(folder_file_dict[folder_name][file_names_key]) 的文件。我有这样的方法：

def process_files_in_series(file_names_array, udp_port):
    for file_name in file_names_array:
         time_consuming_method(file_name, udp_port)
         # create "file_name"

udp_ports = [123, 456, 789]

请注意上面的time_consuming_method()，由于通过 UDP 端口调用需要很长时间。我也仅限于使用上面数组中的 UDP 端口。因此，我必须等待 time_consuming_method 在 UDP 端口上完成，然后才能再次使用该 UDP 端口。这意味着我一次只能运行len(udp_ports) 线程。

因此，我最终将创建len(folder_file_dict.keys()) 线程，并通过len(folder_file_dict.keys()) 调用process_files_in_series。我也有一个 MAX_THREAD 计数。我正在尝试使用Queue 和Threading 模块，但我不确定我需要什么样的设计。我如何使用队列和线程以及可能的条件来做到这一点？使用线程池的解决方案也可能会有所帮助。

注意

我并不是想提高读/写速度。我正在尝试并行化对process_files_in_series 下的time_consuming_method 的调用。创建这些文件只是过程的一部分，而不是速率限制步骤。

另外，我正在寻找使用Queue、Threading 和可能的Condition 模块或与这些模块相关的任何内容的解决方案。线程池解决方案也可能会有所帮助。我不能使用进程，只能使用线程。

我也在寻找 Python 2.7 中的解决方案。

Answer 1

这是如何使用 multiprocessing.Process 的蓝图 使用 JoinableQueue 将作业交付给工人。你会 仍然受 I/O 约束，但使用 Process 你确实有真正的并发， 这可能被证明是有用的，因为线程甚至可能比 处理文件的普通脚本。

（请注意，这也会阻止您使用笔记本电脑做任何其他事情 如果你敢一次启动太多进程:P)。

我试图解释代码 尽可能使用 cmets。

import traceback

from multiprocessing import Process, JoinableQueue, cpu_count

# Number if CPU's on your PC
cpus = cpu_count()

# The Worker Function. Could also be modelled as a class
def Worker(q_jobs):
    while True:
        # Try / Catch / finally may be necessary for error-prone tasks since the processes 
        # may hang forever if the task_done() method is not called.
        try:
            # Get an item from the Queue
            item = q_jobs.get()

            # At this point the data should somehow be processed

        except:
            traceback.print_exc()
        else:
            pass

        finally:
            # Inform the Queue that the Task has been done
            # Without this. The processes can not be killed
            # and will be left as Zombies afterwards
            q_jobs.task_done()


# A Joinable Queue to end the process
q_jobs = JoinableQueue()

# Create process depending on the number of CPU's
for i in range(cpus):

    # target function and arguments
    # a list of multiple arguments should not end with ',' e.g.
    # (q_jobs, 'bla')
    p = Process(target=Worker,
                args=(q_jobs,)
                )
    p.daemon = True
    p.start()

# fill Queue with Jobs
q_jobs.put(['Do'])
q_jobs.put(['Something'])

# End Process
q_jobs.join()

干杯

编辑

我写这篇文章时考虑到了 Python 3。 从 print 函数中取出括号

print item

应该使这项工作适用于 2.7。

Answer 2

使用线程池：

#!/usr/bin/env python2
from multiprocessing.dummy import Pool, Queue # thread pool

folder_file_dict = {
    folder_name: {
        file_names_key: file_names_array
    }
}

def process_files_in_series(file_names_array, udp_port):
    for file_name in file_names_array:
         time_consuming_method(file_name, udp_port)
         # create "file_name"
         ...

def mp_process(filenames):
    udp_port = free_udp_ports.get() # block until a free udp port is available
    args = filenames, udp_port
    try:
        return args, process_files_in_series(*args), None
    except Exception as e:
        return args, None, str(e)
    finally:
        free_udp_ports.put_nowait(udp_port)

free_udp_ports = Queue() # in general, use initializer to pass it to children
for port in udp_ports:
    free_udp_ports.put_nowait(port)
pool = Pool(number_of_concurrent_jobs) #
for args, result, error in pool.imap_unordered(mp_process, get_files_arrays()):
    if error is not None:
       print args, error

如果不同文件名数组的处理时间可能不同，我认为您不需要将线程数绑定到 udp 端口​​数。

如果我正确理解folder_file_dict 的结构，则生成文件名数组：

def get_files_arrays(folder_file_dict=folder_file_dict):
    for folder_name_dict in folder_file_dict.itervalues():
        for filenames_array in folder_name_dict.itervalues():
            yield filenames_array

Answer 3

使用multiprocessing.pool.ThreadPool。它为您处理队列/线程管理，并且可以轻松更改为进行多处理。

编辑：添加示例

这是一个示例...多个线程可能最终使用相同的 udp 端口​​。我不确定这对你来说是否有问题。

import multithreading
import multithreading.pool
import itertools

def process_files_in_series(file_names_array, udp_port):
    for file_name in file_names_array:
         time_consuming_method(file_name, udp_port)
         # create "file_name"

udp_ports = [123, 456, 789]

folder_file_dict = {
         folder_name : {
                         file_names_key : [file_names_array]
                       }
        }

def main(folder_file_dict, udp_ports):
    # number of threads - here I'm limiting to the smaller of udp_ports,
    # file lists to process and a cap I arbitrarily set to 4
    num_threads = min(len(folder_file_dict), len(udp_ports), 4)
    # the pool
    pool = multithreading.pool.ThreadPool(num_threads)
    # build files to be processed into list. You may want to do other
    # Things like join folder_name...
    file_arrays = [value['file_names_key'] for value in folder_file_dict.values()]
    # do the work
    pool.map(process_files_in_series, zip(file_arrays, itertools.cycle(udp_ports))
    pool.close()
    pool.join()