问题的目的: 了解在Python中实现并发的方式/进行实验。
背景: 我想统计匹配特定模式的所有文件中的所有单词数。我的想法是,我可以调用函数 count_words('/foo/bar/*.txt')
,并且所有单词(即由一个或多个空格字符分隔的字符串)将被计算。
在实现中,我正在寻找使用并发实现count_words
的方法。到目前为止,我已经成功地使用了multiprocessing
和asyncio
。
您是否看到完成相同任务的其他替代方法?
由于Python GIL的限制,我没有使用threading
,因为我注意到性能提升并不那么显著。
import asyncio
import multiprocessing
import time
from pathlib import Path
from pprint import pprint
def count_words(file):
with open(file) as f:
return sum(len(line.split()) for line in f)
async def count_words_for_file(file):
with open(file) as f:
return sum(len(line.split()) for line in f)
def async_count_words(path, glob_pattern):
event_loop = asyncio.get_event_loop()
try:
print("Entering event loop")
for file in list(path.glob(glob_pattern)):
result = event_loop.run_until_complete(count_words_for_file(file))
print(result)
finally:
event_loop.close()
def multiprocess_count_words(path, glob_pattern):
with multiprocessing.Pool(processes=8) as pool:
results = pool.map(count_words, list(path.glob(glob_pattern)))
pprint(results)
def sequential_count_words(path, glob_pattern):
for file in list(path.glob(glob_pattern)):
print(count_words(file))
if __name__ == '__main__':
benchmark = []
path = Path("../data/gutenberg/")
# no need for benchmark on sequential_count_words, it is very slow!
# sequential_count_words(path, "*.txt")
start = time.time()
async_count_words(path, "*.txt")
benchmark.append(("async version", time.time() - start))
start = time.time()
multiprocess_count_words(path, "*.txt")
benchmark.append(("multiprocess version", time.time() - start))
print(*benchmark)
为了模拟大量文件,我从古腾堡计划(http://gutenberg.org/)下载了一些书籍,并使用以下命令创建了几个相同文件的副本。
for i in {000..99}; do cp 56943-0.txt $(openssl rand -base64 12)-$i.txt; done
aiofiles
。我可以请您在答案中添加一些相关的更改代码吗? 此外,我以为event_loop.run_until_complete
同时运行函数。当运行代码时,与纯顺序方法相比,我明显看到了差异。谢谢。 - Michael