我最近开始尝试多处理以加快任务速度。我创建了一个脚本，该脚本执行模糊字符串匹配，并使用不同的算法来计算分数(我想比较不同的匹配技术)。您可以在这里找到完整的源代码：https://bitbucket.org/bergonzzi/fuzzy-compare/src。作为输入，它需要将2个文件组合成对(file1的每一行与file2的每一行)。对于每对，计算模糊匹配分数。

我做了3个版本。使用我的存储库中提供的样本数据(合并成对后包含697.340个项目)运行，我有以下计时：

简单的单一流程-0:00:47

使用Pool.map()进行多进程-0:00:13

使用队列的多进程(生产者/消费者模式)-0:01:04

我试图理解为什么Pool.map()版本比Queue版本要快得多，而Queue版本实际上比简单的单进程版本要慢。

我什至尝试使用Queue的理由是，Pool.map()版本会一直保留结果，直到一切都完成了，并且只在最后写入文件。这意味着对于大文件，它最终会占用大量内存。我正在谈论这个版本(链接到它，因为要在此处粘贴很多代码)。

为了解决这个问题，我将其重构为生产者/消费者模式(或至少尝试过)。在这里，我首先通过组合两个输入文件来制作作业，并将它们放入消费者处理的队列(计算模糊匹配分数)。完成的作业将放入出队列。然后，我只有一个过程从此队列中抓取已完成的项目并将它们写入文件。这样，从理论上讲，我不需要那么多的内存，因为结果将被刷新到磁盘上。似乎工作正常，但速度慢得多。我还注意到，在Mac OSX上查看"活动监视器"时，我正在生成的4个进程似乎并没有用完100％的CPU(Pool.map()版本不是这种情况)。

我注意到的另一件事是，我的生产者功能似乎正确填充了队列，但是使用者进程似乎要等到队列被填充，而不是等到第一项到达后才开始工作。我可能在那儿做错了...

作为参考，这里是Queue版本的一些相关代码(尽管最好查看上面链接中的repo中的完整代码)。

这是我的生产者函数：

def combine(list1, list2):

'''

Combine every item of list1 with every item of list 2,

normalize put the pair in the job queue.

'''

pname = multiprocessing.current_process().name

for x in list1:

for y in list2:

# slugify is a function to normalize the strings

term1 = slugify(x.strip(), separator=' ')

term2 = slugify(y.strip(), separator=' ')

job_queue.put_nowait([term1, term2])

这是writer函数：

def writer(writer_queue):

out = open(file_out, 'wb')

pname = multiprocessing.current_process().name

out.write(header)

for match in iter(writer_queue.get,"STOP"):

print("%s is writing %s") % (pname, str(match))

line = str(';'.join(match) + '

')

out.write(line)

out.close()

这是执行实际计算的辅助函数(从代码中摘录了大部分代码，因为在这里没有什么区别，完整的源代码在仓库中)：

def score_it(job_queue, writer_queue):

'''Calculate scores for pair of words.'''

pname = multiprocessing.current_process().name

for pair in iter(job_queue.get_nowait,"STOP"):

# do all the calculations and put the result into the writer queue

writer_queue.put(result)

这是我设置流程的方式：

# Files

to_match = open(args.file_to_match).readlines()

source_list = open(args.file_to_be_matched).readlines()

workers = 4

job_queue = multiprocessing.Manager().Queue()

writer_queue = multiprocessing.Manager().Queue()

processes = []

print('Start matching with"%s", minimum score of %s and %s workers') % (

args.algorithm, minscore, workers)

# Fill up job queue

print("Filling up job queue with term pairs...")

c = multiprocessing.Process(target=combine, name="Feeder", args=(to_match, source_list))

c.start()

c.join()

print("Job queue size: %s") % job_queue.qsize()

# Start writer process

w = multiprocessing.Process(target=writer, name="Writer", args=(writer_queue,))

w.start()

for w in xrange(workers):

p = multiprocessing.Process(target=score_it, args=(job_queue, writer_queue))

p.start()

processes.append(p)

job_queue.put("STOP")

for p in processes:

p.join()

writer_queue.put("STOP")

我在这里已经读到了很多有关多处理有时会变慢的信息??，我知道这与创建和管理新流程的开销有关。同样，当要完成的工作不够"庞大"时，可能看不到多处理的效果。但是在这种情况下，我认为工作量很大，而且Pool.map()版本似乎证明了这一点，因为它的速度要快得多。

在管理所有这些进程并传递队列对象时，我做错什么了吗？如何对其进行优化，以便在处理结果时将其写入文件，以最大程度地减少运行文件时所需的内存量？

谢谢！

我不知道基于队列的系统的性能如何(我看起来还不是很辛苦)，但是要解决基于pool版本的内存消耗问题，您可以尝试使用pool.imap来获得迭代器收益由工作进程计算的结果值。在您的map代码中，只需将imap换为map，然后将pool.close和pool.join调用移到写出结果的循环下，您可能会被设置！

非常感谢您的建议，您的解决方案确实解决了内存问题。使用imap，每个进程占用的空间不到100mb，使用map时，每个进程的空间高达1GB。但是它变慢了-在我的测试数据上，使用imap可以得到34秒(而使用map则需要13秒)。知道为什么会这样吗？

它可能与chunksize参数有关。如果我理解正确，默认情况下map使用的块比imap大，如果要映射的序列很长，则可以大大减少开销。

我如何找到理想的块大小来优化性能？谢谢！

ID建议您在不耗尽内存的情况下使用尽可能大的块大小。当然，最大的收益可能会提早出现(从chunksize=1的默认值增加到chunksize=100)，但是开销越大，开销就应该越少。 map方法计算与math.ceil(len(iterable) 4.0 cpu_count()))等效的默认块大小，因此也可以从imap开始。

通过测试更大的数据，我发现在速度和内存方面，最大为100的块大小(至少对于此用例而言)提供了最佳结果。额外的进程仅需3MB。使用您建议的公式，每个进程占用太多内存(同样，在我的用例中)。尽管主要过程仍然使用很多(接近1GB)，但这可能是因为Im计算了将近一千万个项目的大列表并将其保留在内存中。我仍然找不到通过懒惰的生成器而不是简单列表使用map()的方法...

我认为您的计时问题是您的多线程队列版本缺少优化。您发表的评论实质上是在工作线程开始从中获取作业之前，您的job_queue已填满。我相信这样做的原因是您在#Fill up job queue中具有c.join()。这样可以防止主线程继续执行直到作业队列已满。我将c.join()移到p.join()之后。您还需要找出一种将停止标志放入队列末尾的方法。合并功能可能是放置此内容的好地方。在数据用完后要合并后，添加x个停止标志的方法。

需要注意的另一件事：

您正在开始于p进程的for循环范围内覆盖w变量。作为样式/可读性/等问题，我将w更改为其他变量名称。如果您不使用它，则下划线可以作为一个很好的一次性变量名。

即

for w in xrange(workers):

应该成为

for _ in xrange(workers):

长话短说，如果将c.join()移至末尾，则应该获得更准确的计时。当前，唯一的多线程处理是字符串的模糊匹配。拥有生产者/消费者线程的优点之一是，消费者线程不必等到生产者线程完成后，因此最终将使用较少的内存。

谢谢你的提示！最后加入"求职者"程序是有道理的……我想我仍然对join()实际所做的事情感到困惑，现在对我来说很清楚。我仍然不太了解您的建议："您还需要找出一种将停止标志放入队列末尾的方法。Combine函数可能是放置此标记的好地方。类似添加x的方法数据用完后停止标志的数量。" -你到底是什么意思？为什么x个停止标志？以及如何检测到我没有其他数据要合并？

由于您的数据来自文件，因此只要文件用完，便知道您没有其他数据可以合并。您的使用者线程将需要某种信号来知道何时关闭，从而实现"停止"标志。每当使用者线程将作业从队列中拉出且其状态为" STOP"或"无"或类似情况(您必须选择最适合您的应用程序的线程)时，线程便知道它可以关闭。我如何在类似的应用程序上执行此操作，这是我的生产者线程，一旦完成创建作业，就会产生等于使用者线程数量的停止标志。