multiprocessing模块
进程对象
-
创建
- p = Process(target=foo, args=(param,))
-
属性
- p.daemon: True为守护进程, 守护进程内无法再开启子进程,否则抛出异常:AssertionError: daemonic processes are not allowed to have children, 父进程结束则自己也立刻结束; False则为非守护进程, 自身进程运行与父进程是否结束无关; p.daemon = True | False 必须在p.start()之前调用
- p.name: 进程名
- p.pid: 进程pid; 如果当前进程为父进程, 则p.pid与os.getpid()的结果一样
- p.exitcode: 为None表示进程正在运行, 为-n表示由于某一个信号结束了
- p.start(): 启动一个进程, 内部会调用p.run()方法
- p.join(): 调用p.join()语句的进程需要等待p进程结束才能继续执行, p.join()中会调用wait函数
- p.terminate(): 强制终止p进程, 但是不会立即终止, 所以如果在p.terminate()后紧接着是p.is_alive()则返回True, 但是如果紧接着会后面再来一个p.is_alive()就会返回False了, 第一个p.is_alive()会催促p结束
- p.is_alive(): p进程是否还在运行
-
僵尸进程与进程
-
僵尸进程
- 父进程还在执行, 但是子进程结束了, 父进程没有调用wait或者waitpid函数回收子进程的资源导致子进程死亡了但是仍然占用着进程资源
- 对系统有害, 会造成资源浪费
- 解决
- 如果已经产生了僵尸进程: 杀掉父进程, 让子进程成为孤儿进程从而交给init进程管理即可
- 预防僵尸进程: p为进程对象, 父进程p调用join()方法
- 解决
-
孤儿进程
- 父进程结束, 但是子进程还在执行
- 孤儿进程对系统无害, 孤儿进程会被init进程管理
-
-
注意: Windows与类Unix系统创建子进程的方式不同
-
类Unix
- 类Unix系统采用fork()系统调用函数创建子进程, 字如其名, 子进程就是父进程的一个副本, 拷贝的过程中cs:ip的指针指向指令的哪个位置也是一致的(这多亏了虚拟内存), 也就是说, 复制出来的子进程不是从头开始执行的, 是从父进程调用fork()函数语句的下一条指令开始执行的, 随着技术的发展, fork()现在采用的是CoW实现
-
Windows
-
Windows上创建子进程的函数为CreateProcess(), 也是字如其名, 是创建一个进程而不是复制一个进程。CreateProcess()函数的API是
BOOL CreateProcessA( LPCSTR lpApplicationName, // 进程要执行的.exe文件名 LPSTR lpCommandLine, // 执行的.exe的命令行参数 LPSECURITY_ATTRIBUTES lpProcessAttributes, LPSECURITY_ATTRIBUTES lpThreadAttributes, BOOL bInheritHandles, DWORD dwCreationFlags, LPVOID lpEnvironment, LPCSTR lpCurrentDirectory, LPSTARTUPINFOA lpStartupInfo, LPPROCESS_INFORMATION lpProcessInformation );
- 我们主要看API的第1和第2个参数, 很明显, CreateProcess()API可以创建一个与当前父进程完全不同的子进程, 因为它接受一个.exe文件的路径, 该路径可以是任何一个.exe文件, 将该.exe文件加载到内存中***CPU从头开始执行代码***, 如果要实现与类Unix中fork()函数类似的功能, CreateProcess()的第一个参数应该为父进程的.exe文件的位置, 这样就创建出来一个与父进程一样的子进程了, 但是刚才说了是***类似***, 肯定有不同, ***类Unix中fork出来的子进程的入口是父进程fork语句的下一条指令, 而CreateProcess是从头开始执行子进程***
-
- Windows底层采用CreateProcess函数创建子进程在Python中出现的问题
- 在main.py中
import time import multiprocessing from multiprocessing import Process def foo(): time.sleep(3) print('this is foo function') p = Process(target=foo) p.start() print('Finish')
- 在命令行执行
python3 main.py
, 报错: 常见了无限个进程 - 分析
-
Windows 底层调用CreateProcess函数创建子进程, 创建的子进程会从头开始执行程序, 对于main.py, 创建一个子进程, 就会再走一遍import time..., 这个时候肯定还会遇到Process(), 没有办法, 在CreateProcess一次, 一次类推, 一直创建子进程
-
解决方案
- 依据Python中主进程的
__name__ == __main__
而子进程__name__ != __main__
规避 - 代码
import time import multiprocessing from multiprocessing import Process def foo(): time.sleep(3) print('this is foo function') if __name__ == '__main__': p = Process(target=foo) p.start() print('Finish')
- 依据Python中主进程的
-
- 在main.py中
-
进程共享
- 一般来说进程中定义的数据是不会共享的, 父进程的数据与子进程中的数据无关, 对于一个一般的全局变量也是不共享的; 在这样势必会导致程序运行效率低下, 在Windows中在不同进程中打印id(an_obj), 显示出来的id是不同的, 因为Windows中的进程实质上不是fork出来的而是CreateProcessAPI产生的, 子进程要从头开始走一遍, 具体内容在上文注意中提到, 但是在类Unix中打印出来的id是一样的, 因为是fork出来的;
进程同步
- 进程之间的数据是不共享的, 但是文件系统, 屏幕等是共享的, 可以共同访问一个文件, 一个屏幕(终端), 所以会产生这些资源的竞争, 为此我们需要控制他们的竞争关系
- 为了控制资源的访问, 诞生了进程锁, 这里很特别, 我们知道两个进程(A与B)之间的资源是独立的, 但是multiprocessing中的对象(Lock, Queue等)在两个进程中内部会有复杂的映射, 目的就是要达到资源共享
- 在main程序中定义了multiprocessing.Lock(), 在main中fork出A和B两个子进程执行一个同一段代码(代码一样, 但是不是同一段代码, 是复制出来的两份独立的代码), lock作为参数传入. A中lock.acquire()时, 按道理来说, 进程A与B是独立的, A中调用了acquire()应该不会影响B, 但是Python内部做了复杂的映射, 当A中lock.acquire()时也会对B中同一段代码上锁; 因为lock在内核空间
IPC
-
Queue(建议多使用Queue)
0. Queue内部有锁机制, 并且支持数据共享- q.put方法用以插入数据到队列中,put方法还有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。
- q.get方法可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常.
- q.empty():调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。
- q.full():调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。
- q.qsize():返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样
-
Pipe
- 与Linux C语言中的Pipe有一些不同, Python中的Pipe更加高级, C中的Pipe只能是父子进程之间进行数据交互, 但是Python中的Pipe除了父子之间还可以是其他的进程之间
- Pipe中一端recv或者send时, 如果所有其他端口都被close(所有涉及到该Pipe的进程)了, 才会抛出EOFError异常
- send(obj)和recv()方法使用pickle模块对对象进行序列化
- Pipe编程中的close(), send(), recv()等操作一定要在fork完了所有的进程执行, 否则会产生很多意想不到的错误
-
Manager(共享数据, 内部没有锁, 需要自己加锁)
- 就是C中的mmap
- 一般配合lock与with使用, 因为Manger自己不会加锁
lock = Lock() with Manager() as m: d = m.dict({'data': 100}) p = Process(target=foo, args=(d, lock)) p.start()
-
Semaphore(信号量)
- 信号量规定了一个资源最多最多可以被多少个进程访问, 多出来的会被阻塞
- Semaphore(size)
- acquire() 锁资源
- release() 释放资源锁
-
Pool
- Pool(3), 创建一个有3个进程的进程池, 从无到有, 最多有3个, 之后就一直是3个
- apply 与 apply_async
- apply
- 同步执行进程, 会阻塞当前主进程
- 会立即返回进程执行的结果
- apply_async
- 异步执行进程, 不会立即返回结果, 不会阻塞
- 返回的对象为ApplyResult, 获取结果只需要调用对象的get方法, 但是调用get方法, 需要调用p.close(), 再调用p.join()方法之后才能取结果, 否则主程序结束了, 进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了, 这与我们直接使用Process不同, 要获取结果get, 建议将所有的进程都执行完再批量查看结果
- apply