在做任何工作之前多处理退出

Question

我继承了一个特定的解析器，它应该解析 10 个文件，每个文件大约 4m 行。

代码是用 Python 2 编写的，我对其进行了更新。

有一个多处理逻辑，我似乎无法开始工作。

from multiprocessing.pool import ThreadPool
import glob

DATADIR = 'home/my_dir/where/all/my/files/are'

def process_file(filepath):
    # read line by line, parse and insert to postgres database. 

def process_directory(dirpath):
    pattern = f'{dirpath}/*dat'  # files have .dat extension. 
    tp = ThreadPool(10)
    for filepath in glob.glob(pattern):
        print(filepath)
        tp.apply_async(process_file, filepath)
    tp.close()
    tp.join()

if __name__ == '__main__':
    process_directory(DATADIR)

我浏览了很多文档和一些类似的问题，但它似乎不起作用。

使用解析器代码，我确实会在控制台上打印出我需要解析的文件的所有路径，但程序仅此而已。

Answer 1

问题在于您调用apply_async 的方式。我为您的问题制作了一个简单的复制器，但稍作调整即可从每次调用中获取结果：

from multiprocessing.pool import ThreadPool

def func(f):
    print("hey " + f)
    return f + "1"

l = ["name", "name2", "name3"]
pool = ThreadPool(3)
out = []
for a in l:
    print(a)
    out.append(pool.apply_async(func, a))

# Check the response from each `apply_async` call
for a in out:
    a.get()

pool.close()
pool.join()

这会返回一个错误：

Traceback (most recent call last):
  File "a.py", line 16, in <module>
    a.get()
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 599, in get
    raise self._value
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
TypeError: func() takes 1 positional argument but 4 were given

它认为你传递了四个位置参数，而不是一个。这是因为apply_async 想要在一个元组中传递所有参数，如下所示：

pool.apply_async(func, (a,))

如果你在调用apply_async 时将filepath 放在一个元组中，我想你会得到你期望的行为。

还值得注意的是，您的用例非常适合使用pool.map 而不是apply_async，后者更简洁：

pool.map(process_file, glob.glob(pattern))