笔记：python3 并发编程之多进程multiprocessing模块
一、multiprocessing模块
1、multiprocessing模块介绍
由于GIL（全局解释锁）的问题，python多线程并不能充分利用多核处理器。如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。multiprocessing可以给每个进程赋予单独的Python解释器，这样就规避了全局解释锁所带来的问题。与threading.Thread类似，可以利用multiprocessing.Process对象来创建一个进程。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。Python提供了multiprocessing。
multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。
multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
需要再次强调的一点是：与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。
2、process类的介绍
创建进程的类：
multiprocessing .Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动） 强调： 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
参数介绍：
group参数未使用，值始终为None target表示调用对象，即子进程要执行的任务 args表示调用对象的位置参数元组，args=(1,2,'egon',) kwargs表示调用对象的字典,kwargs={'name':'egon','age':18} name为子进程的名称
方法介绍：
p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
属性介绍：
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
3、process类的使用
注意：在windows中Process()必须放到# if __name__ == '__main__':下
Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module. If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources). This is the reason for hiding calls to Process() inside if __name__ == "__main__" since statements inside this if-statement will not get called upon import. 由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。 如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。 这是隐藏对Process（）内部调用的原，使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调
示例：
import multiprocessing import os def run_proc(name): print('Child process {0} {1} Running '.format(name, os.getpid())) if __name__ == '__main__': print('Parent process {0} is Running'.format(os.getpid())) for i in range(5): p = multiprocessing.Process(target=run_proc, args=(str(i),)) print('process start') p.start() p.join() print('Process close')
以上实例输出的结果
Parent process 809 is Running process start process start process start process start process start Child process 0 810 Running Child process 1 811 Running Child process 2 812 Running Child process 3 813 Running Child process 4 814 Running Process close
process单线程：
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 21:48 # @Author : yangyuanqiang # @File : demo1.py from multiprocessing import Process import time def task(msg): print('hello, %s' % msg) time.sleep(1) if __name__ == '__main__': p = Process(target=task, args=('world',)) p.start() if p.is_alive(): print('Process: %s is running' % p.pid) p.join()
以上实例输出的结果
hello, world Process: 878 is running
这段代码的执行过程：在主进程中创建子进程，然后调用start()启动子进程，调用join()等待子进程执行完，再继续执行主进程的整个的执行流程。
控制子进程进入不同阶段的是 start(), join(), is_alive(), terminate(), exitcode() 方法，这些方法只能在创建子进程的进程中执行。
创建：创建进程需要一个 function 和相关参数，参数可以是dictProcess(target=func, args=(), kwargs = {})，name 可以用来标识进程。
关闭：close停止接收新的任务，如果还有任务来，就会抛出异常。 join 是等待所有任务完成。 join 必须要在 close 之后调用，否则会抛出异常。
等待：在UNIX平台上，当某个进程终结之后，该进程需要被其父进程调用wait，否则进程成为僵尸进程(Zombie)。所以在这里，我们调用了Process对象的join()方法 ，实际上等同于wait的作用。
对于多线程来说，由于只有一个进程，所以不存在此必要性。
结束：terminate() 结束工作进程，不再处理未完成的任务。
process多线程：
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 21:47 # @Author : yangyuanqiang # @File : demo1.py from multiprocessing import Process import multiprocessing import time def task1(msg): print('task1: hello, %s' % msg) time.sleep(1) def task2(msg): print('task2: hello, %s' % msg) time.sleep(1) def task3(msg): print('task3: hello, %s' % msg) time.sleep(1) if __name__ == '__main__': p1 = Process(target=task1, args=('one',)) p2 = Process(target=task2, args=('two',)) p3 = Process(target=task3, args=('three',)) start = time.time() p1.start() p2.start() p3.start() print("The number of CPU is:" + str(multiprocessing.cpu_count())) for p in multiprocessing.active_children(): print("child p.name: " + p.name + " p.id: " + str(p.pid)) p1.join() p2.join() p3.join() end = time.time() print('3 processes take %s seconds' % (end - start))
以上实例输出的结果
task1: hello, one The number of CPU is:4 child p.name: Process-3 p.id: 886 child p.name: Process-2 p.id: 885 child p.name: Process-1 p.id: 884 task2: hello, two task3: hello, three 3 processes take 1.0116989612579346 seconds
这里三个进程执行花费约1s，说明程序是并发执行的。对于更多的并发进程，我们可以放到一个循环中进行处理
Lock
当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 21:50 # @Author : yangyuanqiang # @File : demo3.py from multiprocessing import Lock, Process import time def task1(lock, f): with lock: f = open(f, 'w+') f.write('hello ') time.sleep(1) f.close() def task2(lock, f): lock.acquire() try: f = open(f, 'a+') time.sleep(1) f.write('world!') except Exception as e: print(e) finally: f.close() lock.release() if __name__ == '__main__': lock = Lock() fn = './file.txt' start = time.time() p1 = Process(target=task1, args=(lock, fn,)) p2 = Process(target=task2, args=(lock, fn,)) p1.start() p2.start() p1.join() p2.join() end = time.time() print('time cost: %s seconds' % (end - start)) with open(fn, 'r') as f: for x in f.readlines(): print(x)
以上实例输出的结果
time cost: 2.0096089839935303 seconds
hello world!
因为要访问共享文件，先获得锁的进程会阻塞后面的进程，因此程序运行耗时约2s
Semaphore
Semaphore 和 Lock 稍有不同，Semaphore 相当于 N 把锁，获取其中一把就可以执行了。 信号量的总数 N 在构造时传入，s = Semaphore(N)。 和 Lock 一样，如果信号量为0，则进程堵塞，直到信号大于0。Semaphore可用来控制对共享资源的访问数量，例如池的最大连接数。
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 21:51 # @Author : yangyuanqiang # @File : demo4.py from multiprocessing import Semaphore, Process import time def task(s, msg): s.acquire() print('hello, %s' % msg) time.sleep(1) s.release() if __name__ == '__main__': s = Semaphore(2) processes = [] for x in range(8): p = Process(target=task, args=(s, x,)) processes.append(p) start = time.time() for p in processes: p.start() for p in processes: p.join() end = time.time() print('8 process takes %s seconds' % (end - start))
以上实例输出的结果
hello, 0 hello, 1 hello, 3 hello, 2 hello, 4 hello, 5 hello, 6 hello, 7 8 process takes 4.016672134399414 seconds
信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是Dijkstra信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器、文件这样的有限资源
Event
Event用来实现进程间同步通信
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 22:28 # @Author : yangyuanqiang # @File : demo5.py from multiprocessing import Process, Event import time def task1(e, msg): print('task1 is waitting...') e.wait() time.sleep(1) print('hello, %s, e.is_set(): %s' % (msg, e.is_set())) def task2(e, msg): print('task2 is waitting...') e.wait(msg) print('hello, %s, e.is_set(): %s' % (msg, e.is_set())) if __name__ == '__main__': e = Event() p1 = Process(target=task1, args=(e, 1)) p2 = Process(target=task2, args=(e, 2)) p1.start() p2.start() time.sleep(3) e.set() print('main: event is set')
以上实例输出的结果
task1 is waitting... task2 is waitting... hello, 2, e.is_set(): False main: event is set hello, 1, e.is_set(): True
Pool
如果有50个task要执行，但 CPU 只有4核，我们当然可以循环创建50个进程来做这个事情，但这样处理大大增加进程管理和调度的开销。
如果可以只创建4个进程，让它们轮流工作完成任务，不用我们自己去管理具体的进程的创建、销毁和调度，岂不更好。multiprocessing中的 Pool 可以帮助我们做到这一点
多线程异步
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 22:30 # @Author : yangyuanqiang # @File : demo6.py from multiprocessing import Pool import time def task(msg): print('hello, %s' % msg) time.sleep(1) if __name__ == '__main__': pool = Pool(processes=4) for x in range(10): pool.apply_async(task, args=(x,)) pool.close() pool.join() print('processes done.')
注意：这里使用的是apply_async，多个进程异步执行；如果调用async，就变成阻塞版本了
以上实例输出的结果
hello, 1 hello, 0 hello, 2 hello, 3 hello, 4 hello, 5 hello, 6 hello, 7 hello, 8 hello, 9 processes done.
Pool 进程池创建4个进程，不管有没有任务，都一直在进程池中等候。官网的描述：Worker processes within a Pool typically live for the complete duration of the Pool’s work queue。数据来的时候，若有空闲进程，则利用空闲的进程完成任务，直到所有任务完成为止；若没有空闲的进程，则需要等待，直到池中有进程结束
获取进程执行结果
更多的时候，我们不仅需要多进程执行，还需要关注每个进程的执行结果，我们可以通过获取apply_async的返回值得到执行结果
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 22:32 # @Author : yangyuanqiang # @File : demo7.py from multiprocessing import Pool import time def task(msg): print('hello, %s' % msg) time.sleep(1) return 'msg: %s' % msg if __name__ == '__main__': pool = Pool(processes=4) results = [] for x in range(10): ret = pool.apply_async(task, args=(x,)) results.append(ret) pool.close() pool.join() print('processes done, result:') for x in results: print(x.get())
以上实例输出的结果
hello, 0 hello, 1 hello, 2 hello, 3 hello, 4 hello, 5 hello, 6 hello, 7 hello, 8 hello, 9 processes done, result: msg: 0 msg: 1 msg: 2 msg: 3 msg: 4 msg: 5 msg: 6 msg: 7 msg: 8 msg: 9
pool中的map方法
上面我们是通过一个循环往进程池添加任务，Pool提供了更优雅的map方法来管理任务的提交，只需对上面的代码稍作修改
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/21 22:33 # @Author : yangyuanqiang # @File : demo8.py from multiprocessing import Pool import time def task(msg): print('hello, %s' % msg) time.sleep(1) return 'msg: %s' % msg if __name__ == '__main__': pool = Pool(processes=4) results = [] msgs = [x for x in range(10)] results = pool.map(task, msgs) pool.close() pool.join() print('processes done, result:') for x in results: print(x)
以上实例输出的结果
hello, 0 hello, 1 hello, 2 hello, 3 hello, 4 hello, 5 hello, 6 hello, 7 hello, 8 hello, 9 processes done, result: msg: 0 msg: 1 msg: 2 msg: 3 msg: 4 msg: 5 msg: 6 msg: 7 msg: 8 msg: 9
上课代码：
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/19 20:31 # @Author : lingxiangxiang # @File : demon1.py import multiprocessing import time def worker(args, interval): print("start worker {0}".format(args)) time.sleep(interval) print("end worker {0}".format(args)) def main(): print("start main") p1 = multiprocessing.Process(target=worker, args=(1, 1)) p2 = multiprocessing.Process(target=worker, args=(2, 2)) p3 = multiprocessing.Process(target=worker, args=(3, 3)) p1.start() p2.start() p3.start() print("end main") if __name__ == '__main__': main() # p = multiprocessing.Process(target=, args=) # target 指定的是当进程执行时，需要执行的函数 # args 是当进程执行时，需要给函数传入的参数 # 注意： args必须是一个tuple, 特别是当函数需要传入一个参数时 (1,) # p 代表的是一个多进程， # p.is_alive() 判断进程是否存活 # p.run() 启动进程 # p.start() 启动进程，他会自动调用run方法，推荐使用start # p.join(timeout) 等待子进程结束或者到超时时间 # p.terminate() 强制子进程退出 # p.name 进程的名字 # p.pid 进程的pid
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/19 20:31 # @Author : lingxiangxiang # @File : demon1.py import multiprocessing import time def worker(args, interval): print("start worker {0}".format(args)) time.sleep(interval) print("end worker {0}".format(args)) def main(): print("start main") p1 = multiprocessing.Process(target=worker, args=(1, 1)) p2 = multiprocessing.Process(target=worker, args=(2, 2)) p3 = multiprocessing.Process(target=worker, args=(3, 3)) p1.start() p1.join(timeout=0.5) p2.start() p3.start() print("the number of CPU is: {0}".format(multiprocessing.cpu_count())) for p in multiprocessing.active_children(): print("The name of active children is: {0}, pid is: {1} is alive".format(p.name, p.pid)) print("end main") if __name__ == '__main__': main()
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/19 21:27 # @Author : lingxiangxiang # @File : demon3.py import time import multiprocessing def add1(lock, value, number): with lock: print("start add1 number= {0}".format(number)) for i in range(1, 5): number += value time.sleep(0.3) print("number = {0}".format(number)) def add3(lock, value, number): lock.acquire() print("start add3 number= {0}".format(number)) try: for i in range(1, 5): number += value time.sleep(0.3) print("number = {0}".format(number)) except Exception as e: raise e finally: lock.release() pass if __name__ == '__main__': print("start main") number = 0 lock = multiprocessing.Lock() p1 = multiprocessing.Process(target=add1, args=(lock, 1, number)) p3 = multiprocessing.Process(target=add3, args=(lock, 3, number)) p1.start() p3.start() print("end main")
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2018/5/19 21:27 # @Author : lingxiangxiang # @File : demon3.py import time import multiprocessing from multiprocessing import Value, Array, Manager def add1(value, number): print("start add1 number= {0}".format(number.value)) for i in range(1, 5): number.value += value print("number = {0}".format(number.value)) def add3(value, number): print("start add3 number= {0}".format(number.value)) try: for i in range(1, 5): number.value += value print("number = {0}".format(number.value)) except Exception as e: raise e if __name__ == '__main__': print("start main") number = Value('d', 0) p1 = multiprocessing.Process(target=add1, args=(1, number)) p3 = multiprocessing.Process(target=add3, args=(3, number)) p1.start() p3.start() print("end main")