Python多进程（process）（三）子进程与并行

Python实用教程_spiritx的博客-CSDN博客

subprocess 模块

subprocess模块允许你生成新的进程，连接它们的输入、输出、错误管道，并且获取它们的返回码。此模块打算代替一些老旧的模块与功能：

os.system
os.spawn*

run()

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None, **other_popen_kwargs)

• 运行被 args 描述的指令. 等待指令完成, 然后返回一个 CompletedProcess 实例.
• 以上显示的参数仅仅是最简单的一些，下面 常用参数 描述（因此在缩写签名中使用仅关键字标示）。完整的函数头和 Popen 的构造函数一样，此函数接受的大多数参数都被传递给该接口。（timeout, input, check 和 capture_output 除外）。
• 如果 capture_output 设为 true，stdout 和 stderr 将会被捕获。在使用时，内置的 Popen 对象将自动用 stdout=PIPE 和 stderr=PIPE 创建。stdout 和 stderr 参数不应当与 capture_output 同时提供。如果你希望捕获并将两个流合并在一起，使用 stdout=PIPE 和 stderr=STDOUT 来代替 capture_output。
• 可以指定以秒为单位的 timeout，它会在内部传递给 Popen.communicate()。 如果达到超时限制，子进程将被杀掉并等待。 TimeoutExpired 异常将在子进程终结后重新被引发。 在许多平台 API 上初始进程创建本身不可以被打断因此不保证你能看到超时异常直到至少进程创建花费的时间结束后。
• input 参数将被传递给 Popen.communicate() 以及子进程的 stdin。 如果使用此参数，它必须是一个字节序列。 如果指定了 encoding 或 errors 或者将 text 设置为 True，那么也可以是一个字符串。 当使用此参数时，在创建内部 Popen 对象时将自动带上 stdin=PIPE，并且不能再手动指定 stdin 参数。
• 如果 check 设为 True, 并且进程以非零状态码退出, 一个 CalledProcessError 异常将被抛出. 这个异常的属性将设置为参数, 退出码, 以及标准输出和标准错误, 如果被捕获到.
• 如果指定了 encoding 或 error，或者 text 被设为真值，标准输入、标准输出和标准错误的文件对象将使用指定的 encoding 和 errors 或者 io.TextIOWrapper 默认值以文本模式打开。 universal_newlines 参数等同于 text 并且提供了向后兼容性。 默认情况下, 文件对象是以二进制模式打开的。
• 如果 env 不为 None，则它必须是一个为新进程定义环境变量的映射；它们将顶替继承当前进程环境的默认行为被使用。 它会被直接传递给 Popen。 这个映射在任何平台上均可以是字符串到字符串的映射或者在 POSIX 平台上也可以是字节串到字节串的映射，就像是 os.environ 或者 os.environb。

>>>subprocess.run(["ls", "-l"])  # doesn't capture output
CompletedProcess(args=['ls', '-l'], returncode=0)
 
>>>subprocess.run("exit 1", shell=True, check=True)
Traceback (most recent call last):
  ...
subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1
 
>>>subprocess.run(["ls", "-l", "/dev/null"], capture_output=True)
CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0,
stdout=b'crw-rw-rw- 1 root root 1, 3 Jan 23 16:23 /dev/null\n', stderr=b'')

CompletedProcess

class subprocess.CompletedProcess

run() 的返回值, 代表一个进程已经结束.

主要的属性和方法：

Popen()

class subprocess.Popen(args, bufsize=- 1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, group=None, extra_groups=None, user=None, umask=- 1, encoding=None, errors=None, text=None, pipesize=- 1, process_group=None)

• 在一个新的进程中执行子程序。 在 POSIX 上，该类会使用类似于 os.execvpe() 的行为来执行子程序。 在 Windows 上，该类会使用 Windows CreateProcess() 函数。 Popen 的参数如下。
• args 应当是一个程序参数的序列或者是一个单独的字符串或 path-like object。 默认情况下，如果 args 是序列则要运行的程序为 args 中的第一项。 如果 args 是字符串，则其解读依赖于具体平台，如下所述。 请查看 shell 和 executable 参数了解其与默认行为的其他差异。 除非另有说明，否则推荐以序列形式传入 args。

向外部函数传入序列形式参数的一个例子如下:

Popen(["/usr/bin/git", "commit", "-m", "Fixes a bug."])

在 POSIX，如果 args 是一个字符串，此字符串被作为将被执行的程序的命名或路径解释。但是，只有在不传递任何参数给程序的情况下才能这么做。

• 在 POSIX，当 shell=True， shell 默认为 /bin/sh。如果 args 是一个字符串，此字符串指定将通过 shell 执行的命令。这意味着字符串的格式必须和在命令提示符中所输入的完全相同。这包括，例如，引号和反斜杠转义包含空格的文件名。如果 args 是一个序列，第一项指定了命令，另外的项目将作为传递给 shell （而非命令） 的参数对待。也就是说， Popen 等同于:

Popen(['/bin/sh', '-c', args[0], args[1], ...])

• bufsize 将在 open() 函数创建了 stdin/stdout/stderr 管道文件对象时作为对应的参数供应:
  • 0 表示不使用缓冲区（读取与写入是一个系统调用并且可以返回短内容）
  • 1 表示带有行缓冲（仅在 text=True 或 universal_newlines=True 时有用）
  • 任何其他正值表示使用一个约为对应大小的缓冲区
  • 负的 bufsize （默认）表示使用系统默认的 io.DEFAULT_BUFFER_SIZE。

在 3.3.1 版更改: bufsize 现在默认为 -1 表示启用缓冲以符合大多数代码所期望的行为。 在 Python 3.2.4 和 3.3.1 之前的版本中它错误地将默认值设为 0 即无缓冲并且允许短读取。 这是无意的失误并且与大多数代码所期望的 Python 2 的行为不一致。

• executable 参数指定一个要执行的替换程序。这很少需要。当 shell=True， executable 替换 args 指定运行的程序。但是，原始的 args 仍然被传递给程序。大多数程序将被 args 指定的程序作为命令名对待，这可以与实际运行的程序不同。在 POSIX， args 名作为实际调用程序中可执行文件的显示名称，例如 ps。如果 shell=True，在 POSIX， executable 参数指定用于替换默认 shell /bin/sh 的 shell。
• 如果 preexec_fn 被设为一个可调用对象，此对象将在子进程刚创建时被调用。（仅 POSIX）
• 如果 close_fds 为真值，则除 0, 1 和 2 之外的所有文件描述符都将在子进程执行前被关闭。 而当 close_fds 为假值时，文件描述符将遵循它们的可继承旗标，如 文件描述符的继承 所描述的。
• pass_fds 是一个可选的在父子进程间保持打开的文件描述符序列。提供任何 pass_fds 将强制 close_fds 为 True。（仅 POSIX）
• 如果 cwd 不为 None，此函数在执行子进程前会将当前工作目录改为 cwd。 cwd 可以是一个字符串、字节串或 路径类对象。 在 POSIX 上，如果可执行文件路径为相对路径则此函数会相对于 cwd 来查找 executable (或 args 的第一项)。
• 如果 restore_signals 为 true（默认值），则 Python 设置为 SIG_IGN 的所有信号将在 exec 之前的子进程中恢复为 SIG_DFL。目前，这包括 SIGPIPE ，SIGXFZ 和 SIGXFSZ 信号。 （仅 POSIX）
• 如果 start_new_session 为真值则 setsid() 系统调用将在执行子进程之前在子进程中执行。
• 如果 process_group 为非负整数，则 setpgid(0, value) 系统调用将在执行子进程之前在子进程中执行。
• 如果 group 不为 None，则 setregid() 系统调用将于子进程执行之前在下级进程中进行。 如果所提供的值为一个字符串，将通过 grp.getgrnam() 来查找它，并将使用 gr_gid 中的值。 如果该值为一个整数，它将被原样传递。 （POSIX 专属）
• 如果 extra_groups 不为 None，则 setgroups() 系统调用将于子进程之前在下级进程中进行。 在 extra_groups 中提供的字符串将通过 grp.getgrnam() 来查找，并将使用 gr_gid 中的值。 整数值将被原样传递。 （POSIX 专属）
• 如果 user 不为 None，则 setreuid() 系统调用将于子进程执行之前在下级进程中进行。 如果所提供的值为一个字符串，将通过 pwd.getpwnam() 来查找它，并将使用 pw_uid 中的值。 如果该值为一个整数，它将被原样传递。 （POSIX 专属）
• 如果 umask 不为负值，则 umask() 系统调用将在子进程执行之前在下级进程中进行。
• 如果 env 不为 None，则它必须是一个为新进程定义环境变量的映射；它们将顶替继承当前环境的默认行为被使用。 这个映射在任何平台上均可以是字符串到字符串的映射或者在 POSIX 平台上也可以是字节串到字节串的映射，就像是 os.environ 或者 os.environb。
• 如果指定了 encoding 或 errors，或者如果 text 为真值，则文件对象 stdin, stdout 和 stderr 将使用指定的 encoding 和 errors 以文本模式打开，就如上文 常用参数 中所描述的。 universal_newlines 参数等同于 text 且是出于下向兼容性考虑而提供的。 在默认情况下，文件对象将以二进制模式打开。
• 如果给出， startupinfo 将是一个将被传递给底层的 CreateProcess 函数的 STARTUPINFO 对象。 creationflags，如果给出，可以是一个或多个以下标志之一：
  • CREATE_NEW_CONSOLE
  • CREATE_NEW_PROCESS_GROUP
  • ABOVE_NORMAL_PRIORITY_CLASS
  • BELOW_NORMAL_PRIORITY_CLASS
  • HIGH_PRIORITY_CLASS
  • IDLE_PRIORITY_CLASS
  • NORMAL_PRIORITY_CLASS
  • REALTIME_PRIORITY_CLASS
  • CREATE_NO_WINDOW
  • DETACHED_PROCESS
  • CREATE_DEFAULT_ERROR_MODE
  • CREATE_BREAKAWAY_FROM_JOB
• 当 PIPE 被用作 stdin, stdout 或 stderr 时 pipesize 可被用于改变管道的大小。 管道的大小仅会在受支持的平台上被改变（当撰写本文档时只有 Linux 支持）。 其他平台将忽略此形参。

Popen 对象支持通过 with 语句作为上下文管理器，在退出时关闭文件描述符并等待进程:

with Popen(["ifconfig"], stdout=PIPE) as proc:
    log.write(proc.stdout.read())

Popen 对象

Popen()的返回对象，主要方法和属性：

concurrent模块

concurrent.futures 模块提供异步执行可调用对象高层接口。

异步执行可以由 ThreadPoolExecutor 使用线程或由 ProcessPoolExecutor 使用单独的进程来实现。 两者都是实现抽象类 Executor 定义的接口。

Executor 对象

class concurrent.futures.Executor

抽象类提供异步执行调用方法。要通过它的子类调用，而不是直接调用。

submit(fn, /, *args, **kwargs)
调度可调用对象 fn，以 fn(*args, **kwargs) 方式执行并返回一个代表该可调用对象的执行的 Future 对象。

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())

map(func, *iterables, timeout=None, chunksize=1)
类似于 map(func, *iterables) 函数，除了以下两点：

• iterables 是立即执行而不是延迟执行的；
• func 是异步执行的，对 func 的多个调用可以并发执行。

如果 __next__() 被调用且从对 Executor.map() 原始调用 timeout 秒之后其结果还不可用则已返回的迭代器将引发 TimeoutError。 timeout 可以是整数或浮点数。 如果 timeout 未指定或为 None，则不限制等待时间。

如果 func 调用引发一个异常，当从迭代器中取回它的值时这个异常将被引发。

使用 ProcessPoolExecutor 时，这个方法会将 iterables 分割任务块并作为独立的任务并提交到执行池中。这些块的大概数量可以由 chunksize 指定正整数设置。 对很长的迭代器来说，使用大的 chunksize 值比默认值 1 能显著地提高性能。 chunksize 对 ThreadPoolExecutor 没有效果。

在 3.5 版更改: 加入 chunksize 参数。

shutdown(wait=True, *, cancel_futures=False)
当待执行的 future 对象完成执行后向执行者发送信号，它就会释放正在使用的任何资源。 在关闭后调用 Executor.submit() 和 Executor.map() 将会引发 RuntimeError。

如果 wait 为 True 则此方法只有在所有待执行的 future 对象完成执行且释放已分配的资源后才会返回。 如果 wait 为 False，方法立即返回，所有待执行的 future 对象完成执行后会释放已分配的资源。 不管 wait 的值是什么，整个 Python 程序将等到所有待执行的 future 对象完成执行后才退出。

如果 cancel_futures 为 True，此方法将取消所有执行器还未开始运行的挂起的 Future。无论 cancel_futures 的值是什么，任何已完成或正在运行的 Future 都不会被取消。

如果 cancel_futures 和 wait 均为 True，则执行器已开始运行的所有 Future 将在此方法返回之前完成。 其余的 Future 会被取消。

如果使用 with 语句，你就可以避免显式调用这个方法，它将会停止 Executor (就好像 Executor.shutdown() 调用时 wait 设为 True 一样等待):

import shutil
with ThreadPoolExecutor(max_workers=4) as e:
    e.submit(shutil.copy, 'src1.txt', 'dest1.txt')
    e.submit(shutil.copy, 'src2.txt', 'dest2.txt')
    e.submit(shutil.copy, 'src3.txt', 'dest3.txt')
    e.submit(shutil.copy, 'src4.txt', 'dest4.txt')

在 3.9 版更改: 增加了 cancel_futures。

ProcessPoolExecutor

ProcessPoolExecutor 类是 Executor 的子类，它使用进程池来异步地执行调用。 ProcessPoolExecutor 会使用 multiprocessing 模块，这允许它绕过 全局解释器锁 但也意味着只可以处理和返回可封存的对象。
__main__ 模块必须可以被工作者子进程导入。这意味着 ProcessPoolExecutor 不可以工作在交互式解释器中。
从可调用对象中调用 Executor 或 Future 的方法提交给 ProcessPoolExecutor 会导致死锁。

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(), max_tasks_per_child=None)

• 一个 Executor 子类使用最多由 max_workers 个进程组成的进程池异步执行调用。 如果 max_workers 是 None 或未给出，则默认为机器上的处理器数量。如果 max_workers 小于或等于 0 ，则会引发 ValueError 。在 Windows 系统中，max_workers 必须小于或等于 61 ，否则会引发 ValueError 。如果 max_workers 为 None ，则默认情况下最大为 61 ，即使有更多的处理器可用。mp_context 可以是多进程上下文，也可以是 None。它将用于启动工作者。如果 mp_context 为``None`` 或未给出，则使用默认的多进程上下文。
• initializer 是一个可选的可调用对象，它会在每个工作进程启动时被调用；initargs 是传给 initializer 的参数元组。 如果 initializer 引发了异常，则所有当前在等待的任务以及任何向进程池提交更多任务的尝试都将引发 BrokenProcessPool。
• max_tasks_per_child 是指定单个进程在其退出并替换为新工作进程之前可以执行的最大任务数量的可选参数。 在默认情况下 max_tasks_per_child 为 None 表示工作进程将存活与进程池一样长的时间。 当指定了最大数量时，则如果不存在 mp_context 形参则将默认使用 "spawn" 多进程启动方法。 此特性不能兼容 "fork" 启动方法。

import concurrent.futures
import math
 
PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]
 
def is_prime(n):
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
 
    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True
 
def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))
 
if __name__ == '__main__':
    main()

Future 对象

Future 类将可调用对象封装为异步执行。Future 实例由 Executor.submit() 创建。

class concurrent.futures.Future

将可调用对象封装为异步执行。Future 实例由 Executor.submit() 创建，除非测试，不应直接创建。

主要的方法和属性有：

wait()

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

等待由 fs 指定的 Future 实例（可能由不同的 Executor 实例创建）完成。 重复传给 fs 的 future 会被移除并将只返回一次。 返回一个由集合组成的具名 2 元组。 第一个集合的名称为 done，包含在等待完成之前已完成的 future（包括正常结束或被取消的 future）。 第二个集合的名称为 not_done，包含未完成的 future（包括挂起的或正在运行的 future）。
timeout 可以用来控制返回前最大的等待秒数。 timeout 可以为 int 或 float 类型。 如果 timeout 未指定或为 None ，则不限制等待时间。
return_when 指定此函数应在何时返回。它必须为以下常数之一:

as_completed()

concurrent.futures.as_completed(fs, timeout=None)

返回一个包含 fs 所给出的 Future 实例（可能由不同的 Executor 实例创建）的迭代器，这些实例会在完成时产生 future 对象（包括正常结束或被取消的 future 对象）。 任何由 fs 给出的重复的 future 对象将只被返回一次。 任何在 as_completed() 被调用之前完成的 future 对象将优先被产生。 如果 __next__() 被调用并且在对 as_completed() 的原始调用 timeout 秒之后其结果仍不可用则所返回的迭代器将引发 TimeoutError。 timeout 可以为整数或浮点数。 如果 timeout 未指定或为 None，则不限制等待时间。