我正在使用多个线程处理目录中的所有文件以并行处理文件.一切正常,除了线程似乎保持活着,因此进程的线程数一直上升,直到它达到1K左右的线程,然后它抛出一个thread.error无法启动新的线程错误.我知道这个错误是由线程数的操作系统级别限制引起的.
我似乎无法弄清楚错误的位置是保持线程活着.任何的想法？这是我的代码的最小版本.

class Worker(Thread):
    def __init__(self, tasks):
        Thread.__init__(self)
        self.tasks = tasks
        self.daemon = True
        self.start()

def run(self):
    while True:
        func, args, kargs = self.tasks.get()
        try:
            func(*args, **kargs)
        except Exception, e: print e
        self.tasks.task_done()


class ThreadPool:
    def __init__(self, num_threads):
        self.tasks = Queue(num_threads)
        for _ in range(num_threads): Worker(self.tasks)

    def add_task(self, func, *args, **kargs):
        self.tasks.put((func, args, kargs))

    def wait_completion(self):
        self.tasks.join()


def foo(filename)
    pool = ThreadPool(32)
    iterable_data = process_file(filename)

    for data in iterable_data:
        pool.add_task(some_function, data)
    pool.wait_completion()

files = os.listdir(directory)
for file in files:
    foo(file)

解决方法:

您正在为每个文件启动一个包含32个线程的新ThreadPool.如果你有大量的文件,那将是很多线程.并且由于CPython中一次只能有一个线程执行Python字节码(因为全局解释器锁定),因此它不一定非常快.

将ThreadPool的创建移到foo()函数之外.