Python 并行与并发详解(待更)

Author Avatar
Tr0y 7月 07, 2019 09:51:04 本文共 3.9k 字
  • 文为知己者书
  • 在其它设备中阅读本文章

在 Python 中关于并发/并行的实现方式(库)有很多,于是我打算做个总结方便今后查阅。其实查阅文档,别的不好说,对于 Python,最好的方式就是查阅官方文档,除了少了些例子,几乎是完美的,所以本文只会列出常用的方法与属性。

基础概念

并行

并行(parallelism),就是同时执行的意思,所以单线程永远无法达到并行状态,利用多线程和多进程即可。但是 Python 的多线程由于存在著名的 GIL,无法让两个线程真正“同时运行“,所以实际上是无法到达并行状态的。就像 2 个人同时吃 2 个包子,最后两个包子同时被吃完。

并发

并发(concurrency),整体上来看多个任务同时进行,但是一个时间点只有一个任务在执行。就像 1 个人同时吃 2 个包子,一次一边咬一口,最后差不多是两个包子同时被吃完。

多进程

  1. 进程:程序是指令、数据及其组织形式的描述,进程是程序的实体。
  2. 子进程:子进程指的是由另一进程(对应称之为父进程)所创建的进程。
  3. 多进程:实现并行的手段,多进程的数量取决于 CPU 核心的数量。

多线程

  1. 线程:轻量级进程。线程是进程中的一个实体(一个进程至少有一个线程),是被系统独立调度和分派的基本单位(即操作系统能够进行运算调度的最小单位)。
  2. 多线程:实现并发的手段,多线程数量根据实际需要来确定,但是也是有上限的。

线程自己不拥有系统资源,只拥有一点儿在运行中必不可少的资源,但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程,同一进程中的多个线程之间可以并发执行。由于线程之间的相互制约,致使线程在运行中呈现出间断性。线程也有就绪、阻塞和运行三种基本状态。就绪状态是指线程具备运行的所有条件,逻辑上可以运行,在等待处理机;运行状态是指线程占有处理机正在运行;阻塞状态是指线程在等待一个事件(如某个信号量),逻辑上不可执行。每一个程序都至少有一个线程,若程序只有一个线程,那就是程序本身。

实现方式

多进程

fork

Unix/Linux 操作系统提供了一个 fork() 系统调用,它非常特殊。普通的函数,调用一次,返回一次,但是 fork() 调用一次,返回两次 ,因为操作系统自动把当前进程(父进程)复制了一份(子进程),然后,分别在父进程和子进程内返回。子进程永远返回 0,而父进程返回子进程的 ID。这样做的理由是,一个父进程可以 fork 出很多子进程,所以,父进程要记下每个子进程的 ID,而子进程只需要调用 getpid() 就可以拿到父进程的 ID。

而在 Python 中,os 模块封装了常见的系统调用,其中就包括 fork,可以在 Python 程序中轻松创建子进程:

import os

print('Process (%s) start...' % os.getpid())

# Only works on Unix/Linux/Mac:
pid = os.fork()

if pid == 0:
    print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
    print('I (%s) just created a child process (%s).' % (os.getpid(), pid))

结果:

Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.

有了 fork,一个进程在接到新任务时就可以复制出一个子进程来处理新任务,常见的 Apache 服务器就是由父进程监听端口,每当有新的 http 请求时,就 fork 出子进程来处理新的 http 请求。

subprocess

有时候,子进程并不是代码自身,而是一个外部进程。Python 标准库中的 subprocess 库就可以 fork 一个子进程,来运行一个外部的程序,还可以接管子进程的输入和输出。

subprocess 中定义了多个函数,它们以不同的方式创建子进程(call、check_call、check_output、Popen 等等),根据不同需要来选取即可。另外 subprocess 还提供了一些接管标准流(standard stream)和打通管道(pipe)的工具,从而在进程间使用文本通信。

下面的例子演示了如何在 Python 代码中运行命令 nslookup www.python.org,这和命令行直接运行的效果是一样的:

import subprocess

print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)

结果

$ nslookup www.python.org
Server:        192.168.19.4
Address:    192.168.19.4#53

Non-authoritative answer:
www.python.org    canonical name = python.map.fastly.net.
Name:    python.map.fastly.net
Address: 199.27.79.223

Exit code: 0

multiprocessing

如果要编写多进程的服务程序,Unix/Linux 无疑是正确的选择,由于 Windows 没有 fork 调用,上面的代码在 Windows 上无法运行。但是 Python 是跨平台的,自然也应该提供一个跨平台的多进程支持。于是就出现了 multiprocessing

multiprocessing 模块就是跨平台的多进程模块。在 Unix/Linux 下,multiprocessing 模块封装了 fork(),使我们不需要关注 fork() 的细节。由于 Windows 没有 fork 调用,因此,multiprocessing 需要“模拟”出 fork 的效果,父进程所有 Python 对象都必须通过 pickle 序列化再传到子进程去所有。

所以,如果 multiprocessing 在 Windows 下调用失败了,要先考虑是不是 pickle 失败了。

multiprocessing 模块常见的使用方式:
multiprocessing 模块提供了一个 Process 类来代表一个进程对象:

创建进程的类

Process([group [, target [, name [, args [, kwargs]]]]])
描述:

  1. 由该类实例化得到的对象,表示一个子进程中的任务(尚未启动),下面称之为 p

注意:

  1. args:指定的为传给 target 函数的位置参数,是元组的形式。

参数:

  1. group:参数未使用,值始终为 None
  2. target:表示调用对象,即子进程要执行的任务
  3. args:表示调用对象的位置参数元组,如:args=('arg1', )
  4. kwargs:表示调用对象的字典,如:kwargs={'name': 'hexin', 'age': 18}
  5. name:为子进程的名称

方法:

  1. p.start():启动进程,并调用该子进程中的 p.run()
  2. p.run():进程启动时运行的方法,正是它去调用 target 指定的函数,若继承 Process 类来写多进程的话,自定义的类中一定要实现该方法。
  3. p.terminate():强制终止进程 p,不会进行任何清理操作,如果 p 创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。如果 p 还保存了一个锁那么也将不会被释放,进而导致死锁。
  4. p.is_alive():如果 p 仍在运行,返回 True
  5. p.join([timeout]):主线程等待 p 终止(即主线程处于等的状态,而 p 是处于运行的状态)。timeout 是可选的超时时间(超过这个时间,父线程不再等待子线程,继续往下执行),注意,p.join() 只能 join 住 start 开启的进程,而不能 join 住 run 开启的进程

属性:

  1. p.daemon:默认值为 False,如果设为 True,代表 p 为后台运行的守护进程;当 p 的父进程终止时,p 也随之终止,并且设定为 True 后,p 不能创建自己的新进程;必须在 p.start()之前设置
  2. p.name:进程的名称
  3. p.pid:进程的 pid

示例:
注意:在 windows 中使用 Process() 必须放到 if __name__ == '__main__':

下面的例子演示了启动一个子进程并等待其结束:

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
    print('Run child process %s (%s)...' % (name, os.getpid()))

print('Parent process %s.' % os.getpid())
p = Process(target=run_proc, args=('test',))
print('Child process will start.')
p.start()
p.join()
print('Child process end.')

创建子进程时,只需要传入一个执行函数和函数的参数,创建一个 Process 实例,用 start() 方法启动,这样创建进程比 fork() 还要简单。

join() 方法可以等待子进程结束后再继续往下运行,通常用于进程间的同步。

当然,也可以使用类,继承 Process 来创建子进程:

import time
import random
from multiprocessing import Process

class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name
    def run(self):
        print('%s running' %self.name)

        time.sleep(random.randrange(1,5))
        print('%s stop' %self.name)

p1 = MyProcess('1')
p2 = MyProcess('2')
p3 = MyProcess('3')
p4 = MyProcess('4')

p1.start() # start 会自动调用 run
p2.start()
p3.start()
p4.start()
print('主线程')

结果

1 running
2 running
主线程
3 running
4 running
1 stop
4 stop
2 stop
3 stop

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

那么问题来了,开多进程的目的是为了并发,如果有多核,通常有几个核就开几个进程,进程开启过多,效率反而会下降(开启进程是需要占用系统资源的,而且开启多余核数目的进程也无法做到并行),但很明显需要并发执行的任务常常远大于核数,这时我们就可以通过维护一个进程池来控制进程数目,比如 httpd 的进程模式,规定最小进程数和最大进程数等。

当进程数目不大时,可以直接利用 multiprocessing 中的 Process 类手动创建多个进程,如果数量很大,就需要使用进程池。Pool 类可以提供指定数量的进程,供用户调用,当有新的请求提交到 Pool 中时,如果池还没有满,那么就会创建一个新的进程用来执行该请求;但如果池中的进程数已经达到规定最大值,那么该请求就会等待,直到池中有进程结束,就重用进程池中的进程。

参数:

  1. numprocess:要创建的进程数,如果省略,将默认使用 cpu_count()的值
  2. initializer:是每个子进程启动时要执行的可调用对象,默认为 None
  3. initargs:是要传给 initializer 的参数组

方法:

  1. p.apply(func[, args[, kwargs]]):在一个进程池的所有进程中执行 func(*args,**kwargs),然后返回结果。注意:若同时进行 2 apply,则按照顺序执行,所以这个函数是阻塞型的,阻塞的对象是主进程与各个子进程。这函数从 py2.3 以后就不建议使用了
  2. apply_async(func[, args[, kwds[, callback[, error_callback]]]]):在一个进程池的所有进程中执行 func(*args, **kwargs),然后返回结果,此方法的结果是 AsyncResult 类的实例(下面会说)。callback 是可调用对象,接收输入参数。当 func 的结果变为可用时,将立即传递给 callback,callback 禁止执行任何阻塞操作,否则会收到其他异步操作中的结果。error_callback 是在函数执行出错的时候调用。注意,apply_async 不是阻塞型的。
  3. p.map(func, iterable[, chunksize=None]):Pool 类中的 map 方法,与内置的 map 函数用法行为基本一致,它会使进程阻塞直到返回结果。注意,虽然第二个参数是一个迭代器,但是它只在整个队列都就绪后,程序才会运行子进程。注意:此时 func 必须要接受一个参数,参数来源于 iterable 中的每个元素。这个函数是阻塞型的,阻塞的对象是主进程。
  4. map_async(func, iterable[, chunksize[, callback[, error_callback]]])map_async 与 map 的关系同 apply 与 apply_async,即这个这个函数不是阻塞型的。map_async 与 apply_async 都是异步的,所以需要有个回调函数,也就是通过 callback 参数来指定。当然,error_callback 也是一样的
  5. p.close():关闭进程池,防止进一步操作。
  6. p.terminate():结束子进程,不再处理未处理的任务。
  7. p.join():等待所有子进程结束。注意:此方法只能在 close()teminate() 之后调用,即让进程池不再接受新的任务。

其他方法:
apply_async()map_async() 的返回值是 AsyncResul,这个实例具有以下方法

  1. obj.get([timeout]):返回结果,如果有必要则等待结果到达。timeout 参数是可选的。如果在指定时间内还没有到达,将引发异常。如果在子进程中出现了异常,那么异常将在调用此方法时再次被抛出。(侧重于 拿)
  2. obj.wait([timeout]):等待返回的结果到达,timeout 参数是可选的。如果在指定时间内还没有等到,则直接执行后续代码。(侧重于 等)
  3. obj.ready():如果调用完成,返回 True
  4. obj.successful():如果调用完成且没有引发异常,返回 True,如果在结果就绪之前调用此方法,则会引发异常(ValueError: <multiprocessing.pool.ApplyResult object at xxxxx> not ready

例子:

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print('Run task %s (%s)...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('Task %s runs %0.2f seconds.' % (name, (end - start)))

print('Parent process %s.' % os.getpid())
p = Pool(4)
for i in range(5):
    p.apply_async(long_time_task, args=(i,))

print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')

结果

Parent process 669.
Waiting for all subprocesses done...
Run task 0 (671)...
Run task 1 (672)...
Run task 2 (673)...
Run task 3 (674)...
Task 2 runs 0.14 seconds.
Run task 4 (673)...
Task 1 runs 0.27 seconds.
Task 3 runs 0.86 seconds.
Task 0 runs 1.41 seconds.
Task 4 runs 1.91 seconds.
All subprocesses done.

注意输出的结果,task 0123 是立刻执行的,而 task 4 要等待前面某个 task 完成后才执行,这是因为 Pool 的默认大小在我的电脑上是 4,因此,最多同时执行 4 个进程。这是 Pool 有意设计的限制,并不是操作系统的限制。如果改成:p = Pool(5) 就可以同时跑 5 个进程(但是就不是并行了,而是并发)。由于 Pool 的默认大小是 CPU 的核数,如果你不幸拥有 8 核 CPU,你要提交至少 9 个子进程才能看到上面的等待效果(逃)。

又一个例子:

提交任务,并在主进程中拿到结果(之前的 Process 是执行任务,结果放到队列里,现在可以在主进程中直接拿到结果)

from multiprocessing import Pool
import time

def work(n):
    print('开工啦...')
    time.sleep(3)
    return n ** 2

q = Pool()
# 异步 apply_async 用法:如果使用异步提交的任务,主进程需要使用 join,等待进程池内任务都处理完,然后可以用 get 收集结果,否则,主进程结束,进程池可能还没来得及执行,也就跟着一起结束了
res = q.apply_async(work, args=(2,))
q.close()
q.join() #join 在 close 之后调用
print(res.get())

# 同步 apply 用法:主进程一直等 apply 提交的任务结束后才继续执行后续代码
# res = q.apply(work, args=(2,))
# print(res)

结果

开工啦...
4

Pool 对象调用 join() 方法会等待所有子进程执行完毕,调用 join() 之前必须先调用 close(),调用 close() 之后就不能继续添加新的 Process 了。

End

What do you think?

本文标题: Python 并行与并发详解(待更)
原始链接: http://www.tr0y.wang/2019/07/07/Python并行与并发详解/
发布时间: 2019.07.07-09:51
最后更新: 2019.07.07-21:23
版权声明: 本站文章均采用CC BY-NC-SA 4.0协议进行许可。转载请注明出处!