如何在特定功能上实现多处理？答案

【问题标题】：How to implement multiprocessing on a specific function?如何在特定功能上实现多处理？
【发布时间】：2021-06-03 18:34:40
【问题描述】：

我是这个多处理概念的新手。我正在尝试对拼写函数实现多处理以使其运行得更快。我尝试如下，但没有按之前的顺序得到结果，token 这里是标记化句子的巨大列表。

from spellchecker import SpellChecker
from wordsegment import load, segment
from timeit import default_timer as timer
from multiprocessing import Process, Pool, Queue, Manager

def text_similarity_spellings(self, token):
    """Uses spell checker to separate incorrect spellings and correct them"""
    spell = SpellChecker()
    unknown_words = [list(spell.unknown(word)) for word in token]
    known_words = [list(spell.known(word)) for word in token]
    load()
    segmented = [[segment(word) for word in sub] for sub in unknown_words]
    flat_list = list(self.unpacker(segmented))
    new_list = [[known_words[x], flat_list[x]] for x in range(len(known_words))]
    new_list = list(self.unpacker(new_list))
    newlist = [sorted(set(mylist), key=lambda x: mylist.index(x)) for mylist in new_list]
    return newlist

def run_all(self):
    tread_vta = Manager().list()
    processes = []
    arg_split = np.array_split(np.array(token),10)
    arg_tr_cl = []
    finds = []
    trdclean1 = []
    for count, k in enumerate(arg_split):
        arg_tr_cl.append((k, [], tread_vta, token[t]))
    for j in range(len(arg_tr_cl)):
        p = Process(target= self.text_similarity_spellings, args=arg_tr_cl[j])
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

谁能建议我一种更好的方法来将多处理应用于特定函数并以正确的顺序获得结果？

【问题讨论】：

我认为您的应用程序更适合map，特别是如果您关心订单。 Process 更适合异步操作，您不关心结果是否以相同的顺序返回。这是因为单个进程可以更快地完成 // 更慢

标签： python multiprocessing python-multiprocessing spell-checking

【解决方案1】：

首先，创建进程会产生一定的开销，然后将参数从主进程传递到子进程（它“存在”在另一个地址空间中）并取回返回值（顺便说一下，您没有为从工作函数 text_similarity_spellings 实际获取返回值做出任何规定。因此，为了让您从使用多处理中获利，并行执行任务（调用工作函数）所获得的收益必须足以抵消上述额外的成本。所有这些只是一种说法，即您的工作函数必须足够占用 CPU 资源才能证明多处理的合理性。

其次，考虑到创建流程的成本，您不希望创建的流程超出您可能使用的数量。如果你有N 任务要完成（arg_tr_cl 的长度）和M CPU 处理器来运行它们并且你的工作函数是纯 CPU（不涉及 I/O），那么尝试使用超过M 的进程运行这些任务，您永远无法获得任何收益。但是，如果它们确实结合了一些 I/O，那么使用更多进程可能会有利可图。如果涉及大量 I/O 并且只涉及一些 CPU 密集型处理，那么结合使用多线程和多处理可能是可行的方法。最后，如果worker函数主要是I/O，那么多线程就是你想要的。

有一种解决方案可以使用X 进程（基于您确定的X 的任何值）来完成N 任务并且能够从您的工作函数，即使用大小为X 的进程池。

MULTITHREADING = False

n_tasks = len(arg_tr_cl)

if MULTITHREADING:
    from multiprocessing.dummy import Pool

    # To use multithreading instead (we can use a much larger pool size):
    pool_size = min(n_tasks, 100) # 100 is fairly arbitrary

else:
    from multiprocessing import Pool, cpu_count

    # No point in creating pool size larger than the number of tasks we have
    # Otherwise, assuming we are mostly CPU-intensive, just create pool size
    # equal to the number of cpu cores that we have:
    n_processors = cpu_count()
    pool_size = min(n_tasks, n_processors)

pool = Pool(pool_size)
return_values = pool.map(self.text_similarity_spellings, arg_tr_cl)
# You can now iterate return_values to get the return values:
for return_value in return_values:
    ...
# or create a list, for example: return_values = list(return_values)

但如果每次调用都必须读取外部字典，SpellChecker 可能会执行大量 I/O。如果是这种情况，您的最佳性能是否可能是初始化一次SpellChecker，然后循环检查每个单词并完全忘记多处理（或多线程）？

【讨论】：