Python并发编程之进程池

一、进程池简介

可以用Pool类来创建进程池，可以把各种数据处理任务都提交给进程池。进程池提供的功能有点类似于列表解析和功能性编程操作提供的功能。

Pool([numprocess [, initializer [, initargs]]])

numprocess是要创建的进程数，如果省略此参数，将使用cpu_cout()的值。initalizer是每个工作进程启动时要执行的可调用对象。initargs是要传递给initializer的参数元组。initializer默认是none。

Pool类的实例p支持以下操作。

• p.apply(func [, args [, kwargs]])
• p.apply_async(func [, args [, kwargs [, callback]]])
• p.close()
• p.join()
• p.imap(func, iterable [, chunksize])
• p.imap_uinordered(func, iterable [, chunksize])
• p.map(func, iterable [, chunksize])
• p.terminate()

方法apply_async()和map_async()的返回值是AsyncResult实例。AsyncResult实例具有以下方法：

• a.get([timeout])
• a.ready()
• a.successful()
• a.wait([timeout])

二、进程池代码实例

下面的例子说明如何使用进程池构建字典，将整个目录中文件的文件内容映射为SHA512摘要值

import os
import multiprocessing
import hashlib

# some parameters you can tweak
BUFFSIZE = 8192     # 读取缓冲区的大小
POOLSIZE = 2        # 工作进程数量

def compute_digest(filename):
    try:
        f = open(filename, "rb")
    except IOError:
        return None
    digest = hashlib.sha512()
    while True:
        chunk = f.read(BUFFSIZE)
        if not chunk:
            break
        digest.update(filename.encode('utf-8'))
    f.close()
    return filename, digest.hexdigest()

def build_digest_map(topdir):
    digest_pool = multiprocessing.Pool(POOLSIZE)
    all_files = (os.path.join(path, name)
                    for path, dirs, files in os.walk(topdir)
                        for name in files)
    digest_map = dict(digest_pool.imap_unordered(compute_digest, all_files, 20))
    digest_pool.close()
    return digest_map

# 尝试按照需要修改目录名称
if __name__ == '__main__':
    digest_map = build_digest_map('C:\\N-20N3PF2C6RYY-Data\\bihu\\Music')
    print(len(digest_map))
    print(digest_map)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

在这个例子中，使用生成器表达式指定一个目录树中所有文件的路径名称序列，然后使用imap_unordered()函数将这个序列分割并传递给进程池。每个工作进程使用compute_digest()函数为它的文件计算SHA512摘要值。将结果发回给生成器，然后发回给生成器。

只有充分利用池工作进程才能使额外的通信开销变得有价值，使用进程池才有意义。一般而言，对于简单的计算（如两数相加），使用进程池是没有意义的。