AI技术百科
8.9、Python multiprocessing模块精讲
multiprocessing 模块无须安装,从 Python 2.6 开始系统便自带该模块了。该模块的接口函数和 threading 类似,但是它启动的是进程而不是线程。
使用该模块时需要先将其引入 multiprocessing,方法如下:
import multiprocessing
该模块包含很多类,如Lock,其和多线程中的锁类似,本节对这些类都会有所涉及。
创建进程
创建进程最简单的方法是创建一个 multiprocessing.Process 的实例,在创建该实例时需要提供入口函数。
下面就是一个最简单的创建进程的例子:
import time, os import multiprocessing # 引入multiprocessing模块 # 子进程要执行的代码,可以看作是进程的入口函数 def process_entry(): print(u"子进程在运行") print(u'子进程的ID = %d' % os.getpid()) if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) p = multiprocessing.Process(target=process_entry) time.sleep(1) print(u'启动子进程') p.start() # 此时才真正开始运行子进程 time.sleep(2) print(u'父进程结束')
运行结果如下:
$ python multiprocessingDemo1.py # 运行脚本
父进程的ID = 63998
启动子进程
子进程在运行
子进程的ID = 63999
父进程结束
当然创建进程时也可以带上参数。上例子中没有带任何参数,下面的例子中带了一个参数。
import time, os import multiprocessing # 引入multiprocessing模块 # 子进程要执行的代码,入口函数带有一个参数arg1 def process_entry(arg1): # 子进程的入口函数 print(u"子进程(%d)在运行" % arg1) print(u'子进程(%d)的ID = %d' % (arg1, os.getpid())) if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) # 指定了参数为1 p1 = multiprocessing.Process(target=process_entry, args=(1,)) # 指定了参数为2 p2 = multiprocessing.Process(target=process_entry, args=(2,)) time.sleep(1) print(u'启动子进程') # 此时子进程才开始执行 p1.start() p2.start() time.sleep(2) print(u'父进程结束')
运行结果如下:
$ python multiprocessingDemo2.py
父进程的ID = 64055
启动子进程
子进程(1)在运行
子进程(1)的ID = 64056
子进程(2)在运行
子进程(2)的ID = 64057
父进程结束
另外一种创建进程的方法是从 Process 派生一个自己的类,我们只需要定义该类的 run() 函数即可。启动进程时就会运行该 run() 函数。
import time, os import multiprocessing # 用户自建的进程类 class NewProcess(multiprocessing.Process): def __init__(self, arg): # 初始化函数 super(NewProcess, self).__init__() self.arg = arg def run(self): # 入口函数 print(u"子进程(%d)在运行" % self.arg) print(u'子进程(%d)的ID = %d' % (self.arg, os.getpid())) if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) p1 = NewProcess(1) p2 = NewProcess(2) time.sleep(1) print(u'启动子进程') p1.start() # 启动子进程 p2.start() time.sleep(2) # 休眠2秒钟 print(u'父进程结束')
运行结果如下:
$ python3 multiprocessingDemo5.py
父进程的ID = 62560
启动子进程
子进程(1)在运行
子进程(1)的ID = 62561
子进程(2)在运行
子进程(2)的ID = 62562
父进程结束
进程的属性
在创建完进程后,可以对其进行很多操作,如启动、退出、查看其运行状态等。本节将介绍与进程相关的一些属性。
1) 进程ID
这个和操作系统返回的 PID 是一样的。可以通过下面的例子来演示其用法:
import time, os import multiprocessing def child_process_entry(): pid = os.getpid() ppid = os.getppid() print(u"子进程: PID = %d, PPID = %d" % (pid, ppid)) while True: time.sleep(10) main_pid = os.getpid() child_process = multiprocessing.Process(target=child_process_entry) child_process.start() print(u"主进程: PID=%d" % main_pid) print(u"主进程: 子进程的PID=%d" % child_process.pid) child_process.join()
如果是在Linux或者macOS系统上运行,其可能的输出如下:
$ python3 demo2.py # 启动脚本 主进程: PID=52983 # 在主进程中得到自己的PID 主进程: 子进程的PID=52984 # 在主进程中得到子进程的PID # 在子进程中得到子进程的PID和主进程的PID 子进程: PID = 52984, PPID = 52983
该程序是不会自动退出的,因为子进程是一个死循环,而主进程则是一直在等待子进程结束。所以需要强制让其退出,可以通过按 Ctrl+C 组合键或者关闭窗口等方法来强制退出。
我们可以在 shell 中看到这两个进程,如下:
$ ps ax | grep python # 查看所有的Python进程 52983 s001 S+ 0:00.05 python3 demo2.py # 主进程,第一列是PID 52984 s001 S+ 0:00.00 python3 demo2.py # 子进程 53585 s003 S+ 0:00.00 grep python # 这是我们的查询进程,不用管
在这个例子中,使用 os.getpid() 得到当前进程的 PID,使用 os.getppid() 得到当前进程的父进程的 PID。
如果是在 Windows 系统下运行,需要加上下面运行代码:
multiprocessing.freeze_support()
完整代码如下:
import time, os import multiprocessing def child_process_entry(): # 子进程入口 pid = os.getpid() ppid = os.getppid() print(u"子进程: PID = %d, PPID = %d" % (pid, ppid)) while True: time.sleep(10) if __name__ == '__main__': # 主程序 # windows下必须要有的 multiprocessing.freeze_support() # 新加的代码 main_pid = os.getpid() child_process = multiprocessing.Process(target=child_process_entry) child_process.start() print(u"主进程: PID=%d" % main_pid) print(u"主进程: 子进程的PID=%d" % child_process.pid) child_process.join()
运行结果如下:
> python demo2.py
主进程: PID=1008
主进程: 子进程的PID=4432
子进程: PID = 4432, PPID = 1008
此时在 Windows 任务管理器中可以看到两个 Python 进程,如图 1 所示。
图 1 进程ID
可以看到代码返回的 PID 和操作系统中的 PID 是一致的。
另外,属性 ident 也是表示进程 ID 的,它们其实是同一个对象。下面的代码使用 id() 检查并确认它们为同一个对象。
import time, os import multiprocessing def process_entry(arg1): # 子进程要执行的代码 return arg1 * 2 if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) p1.start() if id(p1.pid) == id(p1.ident): print(u"pid和ident是同一个对象") time.sleep(1)
运行结果如下:
$ python multiprocessing_pid_ident.py
pid和ident是同一个对象
需要注意的是,在调用start()之前,该属性的值为 None;在进程退出后,该属性的值依然有效。
2) Daemon属性
Daemon 进程在父进程退出时自动退出,而且不能再创建新的进程。该属性默认值是 False,表示普通的进程。
我们可以在创建进程时通过参数 daemon=True 来创建一个 Daemon 进程。下面先来看看非 Daemon 进程在父进程退出后状态。在下面的例子中,父进程在启动子进程 10 秒钟后就退出了,但是子进程需要运行 15 秒钟才退出。
import time, os import multiprocessing # 子进程要执行的代码 def process_entry(arg1): print(u"子进程(%d)在运行" % arg1) print(u'子进程(%d)的ID = %d' % (arg1, os.getpid())) round = 5 while round > 0: print(u"子进程在运行中") time.sleep(3) round = round - 1 print(u"子进程退出") if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) p1 = multiprocessing.Process(target=process_entry, args=(1,),daemon=False) time.sleep(1) print(u'启动子进程') p1.start() time.sleep(10) print(u'父进程结束')
运行结果如下:
$ python multiprocessingDemo3.py
父进程的ID = 65687
启动子进程
子进程(1)在运行
子进程(1)的ID = 65688
子进程在运行中
子进程在运行中
子进程在运行中
子进程在运行中
父进程结束
子进程在运行
可以看到在父进程退出后,子进程继续执行。
如果在创建时指定子进程是 Daemon 进程,那么在父进程执行完毕后子进程会被强制退出。修改上例中的第 15 行代码,原来是:
p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=False)
现在修改为:
p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True)
运行修改后的代码,可以看到下面的输出:
$ python multiprocessingDemo4.py
父进程的ID = 9734
启动子进程
子进程(1)在运行
子进程(1)的ID = 9735
子进程在运行中
子进程在运行中
子进程在运行中
子进程在运行中
父进程结束
可以看到在父进程退出后子进程自动被强制退出。
我们也可以在调用进程的 start() 之前设置该属性,方法如下:
进程对象. Daemon = True
3) exitcode进程返回码
就是进程函数返回的值。下面的代码演示了如何得到进程的返回码。
import time, os, sys import multiprocessing def process_entry(arg1): # 子进程要执行的代码 sys.exit(arg1*2) if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) p1.start() time.sleep(1) print(u"返回值是%d" % p1.exitcode)
运行结果如下:
$ python multiprocessing_exit_code.py
返回值是2
需要注意的是,需要使用 sys.exit() 来指定返回码,不能使用 return,而且仅在进程结束运行时才可以得到返回码,否则该属性的值为 None。
进程的接口函数
除了可以获得进程的属性,进程实例对象还提供了一些接口函数,通过这些接口函数可以对进程进行操作,如启动进程、判断进程是否仍然在运行、得到进程退出码、等待进程退出、强制要求进程退出等。
1) start():启动
就像前面演示的那样,只有在调用该函数之后,进程才真正运行起来,进程对象也才有了进程的 ID。在调用该函数之前,进程对象的 pid 属性为 None。下面的代码演示这个情况:
import time, os import multiprocessing # 子进程要执行的代码 def process_entry(arg1): print(u"子进程在运行") return arg1 * 2 if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) if p1.pid is None: print(u"1)在调用start()之前,子进程的ID=None") else: print(u"1)在调用start()之前,子进程的ID=%d" % p1.pid) p1.start() time.sleep(1) print(u"2)在调用start()之后,子进程的ID=%d" % p1.pid) time.sleep(1)
运行结果如下:
$ python multiprocessing_noIDbeforeStart.py
1)在调用start()之前,子进程的ID=None
子进程在运行
2)在调用start()之后,子进程的ID=12430
而且对于某一个进程对象,只能调用一次该函数,如果多次调用,除第一次之外,其他的都会抛出异常。例如下面的代码:
import time, os import multiprocessing # 子进程要执行的代码 def process_entry(arg1): print(u"子进程在运行") return arg1 * 2 if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) p1.start() time.sleep(1) print(u'再次启动该进程对象,导致错误') p1.start() # 再次启动,抛出异常
运行结果如下:
$ python multiprocessing_restart.py
子进程在运行
再次启动该进程对象,导致错误
Traceback (most recent call last):
File "multiprocessing_restart.py", line 14, in <module>
p1.start()
File "/anaconda3/lib/python3.7/multiprocessing/process.py", line
106, in start
assert self._popen is None, 'cannot start a process twice'
AssertionError: cannot start a process twice
2) is_alive():进程是否还在运行
在调用 start() 之前,该函数返回 False;在该进程退出后,该函数返回 False;在其他时候,该函数返回 True。
下面的例子演示了该函数在不同时刻返回的值。
import time, os import multiprocessing def process_entry(arg1): # 子进程要执行的代码 max_round = 2 while max_round > 0: time.sleep(3) max_round = max_round - 1 print(u'子进程结束') if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) print(u"1)子进程是否存活: %s" % p1.is_alive()) p1.start() print(u"2)子进程是否存活: %s" % p1.is_alive()) max_query = 10 while max_query > 0: time.sleep(1) print(u"3)子进程是否存活: %s" % p1.is_alive()) max_query = max_query - 1 print(u'父进程结束')
运行结果如下:
$ python multiprocessing_isalive1.py
1)子进程是否存活: False
2)子进程是否存活: True
3)子进程是否存活: True
3)子进程是否存活: True
3)子进程是否存活: True
3)子进程是否存活: True
3)子进程是否存活: True
子进程结束
3)子进程是否存活: False
3)子进程是否存活: False
3)子进程是否存活: False
3)子进程是否存活: False
3)子进程是否存活: False
父进程结束
3) join(超时时间)等待进程结束
该函数在前面的例子中使用过。如果没有指定超时时间则一直等待,直到指定进程退出为止。
import time, os import multiprocessing def process_entry(arg1): # 子进程要执行的代码 print(u"子进程(%d)在运行" % arg1) print(u'子进程(%d)的ID = %d' % (arg1, os.getpid())) max_round = 3 while max_round > 0: print(u"%d) 子进程在运行中" % max_round) time.sleep(3) max_round = max_round - 1 if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) time.sleep(1) print(u'启动子进程') p1.start() p1.join() # 这个函数没有返回值 print(u'父进程结束')
运行结果如下:
$ python multiprocessing_joinDemo1.py
父进程的ID = 9940
启动子进程
子进程(1)在运行
子进程(1)的ID = 9941
3) 子进程在运行中
2) 子进程在运行中
1) 子进程在运行中
父进程结束
如果指定了超时时间,则可能在指定进程退出之前该函数就返回了。这时不能使用返回值来判断是因为进程退出了还是超时了,而需要使用 is_alive() 来判断是否是因为超时而返回的。下面的代码演示了这种用法。
import time, os import multiprocessing def process_entry(arg1): # 子进程要执行的代码 print(u"子进程(%d)在运行" % arg1) print(u'子进程(%d)的ID = %d' % (arg1, os.getpid())) max_round = 3 while max_round > 0: print(u"%d) 子进程在运行中" % max_round) time.sleep(3) max_round = max_round - 1 print(u'子进程结束') if __name__=='__main__': print(u'父进程的ID = %d' % os.getpid()) p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) time.sleep(1) print(u'启动子进程') p1.start() while p1.is_alive(): p1.join(1) # 仅仅等待1秒钟 print(u'父进程结束')
运行结果如下:
$ python multiprocessing_joinDemo2.py
父进程的ID = 9985
启动子进程
子进程(1)在运行
子进程(1)的ID = 9986
3) 子进程在运行中
2) 子进程在运行中
1) 子进程在运行中
子进程结束
父进程结束
4) kill():强制退出
该函数给指定的进程发送 SIGKILL 信号。如果是 Windows 系统,那么就是调用了 TerminateProcess() 接口函数。
另外一个强制退出的接口函数是 terminate(),其内部实现和接口函数 kill() 会有所不同,但我们可以将它们当作同一个接口函数来使用。下面的代码演示了该接口函数的用法。
import time, os, sys import multiprocessing def process_entry(arg1): # 子进程要执行的代码 while True: # 子进程不会自主,是一个死循环 time.sleep(1) if __name__=='__main__': p1 = multiprocessing.Process(target=process_entry, args=(1,), daemon=True) p1.start() if p1.is_alive(): print(u"子进程在运行中") else: print(u"子进程没有在运行中") print(u"杀死子进程") p1.kill() time.sleep(2) if p1.is_alive(): print(u"子进程在运行中") else: print(u"子进程没有在运行中")
运行该程序,输出如下:
$ python multiprocessing_kill.py
子进程在运行中
杀死子进程
子进程没有在运行中
需要注意的是,并不是所有的子进程都可以被杀死掉的,也不是立即就可以被杀死掉的。就像杀死 Windows
下的某些进程,由于这些进程忽略对外部消息的响应,导致它们很难被杀死。我们也不认为从外部杀死一个进程是一个好的想法。相对来说,让进程自主优雅地退出是一个更好的设计。
进程池
如果频繁去创建一个进程,然后销毁它,会导致性能下降。对于这种情况推荐的做法是事先创建一个进程池,在有任务达到时从进程池中取出一个进程来执行相关任务,在任务完成后便归还回去。这样做可以复用部分已有进程资源,达到提升效率的作用。
在 mulitprocessing 模块中有一个 Pool 类可以帮助我们完成该任务。下面是一个使用进程池的例子。
import multiprocessing # 引入multiprocessing模块 import time # 引入time模块 def child_process_entry(): # 定义进程入口函数 print(u"子进程在运行") time.sleep(10) print(u"子进程结束") pool_obj = multiprocessing.Pool(processes = 5) # 建立5个元素池子 for i in range(10): # 添加10个进程 pool_obj.apply_async(child_process_entry) pool_obj.close() # 停止添加进程了 pool_obj.join() # 等待进程都结束
运行该脚本后,输出如下:
$ python3 process_pool1.py # 运行脚本
子进程在运行 # 同时启动5个进程来完成任务
子进程在运行
子进程在运行
子进程在运行
子进程在运行
子进程结束 # 有进程结束工作了
子进程结束
子进程结束
子进程结束
子进程结束
子进程在运行 # 启动新的任务
子进程在运行
子进程在运行
子进程在运行
子进程在运行
子进程结束
子进程结束
子进程结束
子进程结束
子进程结束
可以发现,其一次启动 5 个进程,并在 5 个完成后才启动另外 5 个。也就是说其最多同时运行 5 个进程,并且仅当有进程结束后才将新的进程投入运行。
启动进程时可以指定参数 args,args 是一个元组,其各个元素对应到入口函数的各个参数。下面将前面的例子稍作修改,传入一个参数,用来标识该进程。具体代码如下:
import multiprocessing # 引入multiprocessing模块 import time, random # 引入time,random模块 def child_process_entry(): # 定义进程入口函数 print(u"子进程在运行") time.sleep(random.randint(1, 100)/10.0) # 休息随时间 print(u"子进程结束") pool_obj = multiprocessing.Pool(processes = 5) # 建立5个元素池子 for i in range(10): # 添加10个进程 pool_obj.apply_async(child_process_entry) pool_obj.close() # 停止添加进程了 pool_obj.join() # 等待进程都结束
运行该脚本,输出如下:
$ python3 process_pool1.py # 运行脚本
子进程在运行 # 同时启动5个进程来完成任务
子进程在运行
子进程在运行
子进程在运行
子进程在运行
子进程结束 # 有进程结束工作了
子进程在运行 # 投入一个新的进程
子进程结束 # 有进程结束工作了
子进程在运行 # 投入一个新的进程
子进程结束 # 有进程结束工作了
子进程在运行 # 投入一个新的进程
子进程结束 # 有进程结束工作了
子进程在运行 # 投入一个新的进程
子进程结束 # 有进程结束工作了
子进程在运行 # 投入一个新的进程
子进程结束
子进程结束
子进程结束
子进程结束
子进程结束
由于是多进程同时运行,所以一般子进程和主进程是同时运行的,也就是说它们是异步的,这也是 apply_async 中 async 的来源。
如果要是同步执行,那么主进程会等待子进程结束,所以子进程只能是一个一个地运行。这种同步运行的方式比较少见,因为这没有利用到并发操作的特性。但这里还是要演示一下同步运行的情况,将上面的代码稍作修改,将第9行从
pool_obj.apply_async(child_process_entry, args=(i, ))
修改为
pool_obj.apply(child_process_entry, args=(i, ))
其他部分不做任何修改,运行后的输出如下:
$ python3 process_pool3.py
子进程0在运行 # 第一个进程开始执行
子进程0结束 # 第一个进程执行完毕
子进程1在运行
子进程1结束
子进程2在运行
子进程2结束
子进程3在运行
子进程3结束
子进程4在运行
子进程4结束
子进程5在运行
子进程5结束
子进程6在运行
子进程6结束
子进程7在运行
子进程7结束
子进程8在运行
子进程8结束
子进程9在运行
子进程9结束
可以看到子进程是依次执行的,前一个执行完毕之后才执行下一个。
进程通信
在前面章节我们也用到了进程之间的一些简单信息交换,如查询进程的运行状态、等待进程退出、得到进程的退出码等。但是在一些复杂的应用场景,则希望进程之间有更多的信息交换。下面介绍 multiprocessing 模块提供的进程之间通信的方式,包括管道、队列和锁。
1) 管道
管道有两头,一般一头给进程 A,一头给进程 B。如果进程 A 对管道进行写入操作,那么进程B就可以通过读操作看到写入的数据。而且管道是双向的,可以进程 A 写进程 B 读,也可以进程 B 写进程 A 读。
下面的例子演示了管道的用法。
import multiprocessing # 引入multiprocessing模块 def process_A(pipe): # 进程A的入口函数 print(u"进程A发送数据hello到B") # 发送数据hello到进程B pipe.send('hello') print(u"进程A等待进程B的输入") data = pipe.recv() # 结束进程B的数据 print(u'进程A收到B的数据%s' % data) print(u'进程A结束') def process_B(pipe): # 进程B的入口函数 print(u'进程B等待进程A的数据') data = pipe.recv() # 接收数据 print(u'进程B收到B的数据%s' % data) print(u'进程B发送hi到进程A') pipe.send('hi') # 发送数据 print(u'进程B结束') pipe = multiprocessing.Pipe() # 建立管道 # 创建进程A p1 = multiprocessing.Process(target=process_A, args=(pipe[0],)) # 创建进程B p2 = multiprocessing.Process(target=process_B, args=(pipe[1],)) p1.start() # 启动进程 p2.start() p1.join() # 等待进程结束 p2.join()
运行后的输出如下:
$ python3 pipe1.py
进程A发送数据hello到B
进程A等待进程B的输入
进程B等待进程A的数据
进程B收到B的数据hello
进程B发送hi到进程A
进程B结束
进程A收到B的数据hi
进程A结束
2) 队列
对于队列有两个操作,一个是写入数据,一个是读出数据。最简单的使用方式是让一个进程往队列中写入数据,让另外一个进程从队列中读出数据。
import multiprocessing import time def process_read(queue): # 读队列子进程 print(u"queue读出进程开始运行") data = queue.get() while(data != "quit"): # 一直读,直到读出了quit print("queue读出进程读出数据%s" % data) data = queue.get() print(u'queue读出进程退出') def process_write(queue): # 写队列子进程 print(u"queue写入进程开始运行") data = ['good', 'morning', 'everyone'] # 写入的数据 for w in data: print("queue写入进程写入数据%s" % w) data = queue.put(w) time.sleep(1) data = queue.put('quit') # 发送quit,让读子进程退出 print(u'queue写入进程退出') queue_obj = multiprocessing.Queue(3) # 创建两个子进程,一个读队列,一个写队列 p1 = multiprocessing.Process(target=process_read, args=(queue_obj,)) # Pass the other end of the pipe to process 2 p2 = multiprocessing.Process(target=process_write, args=(queue_obj,)) p1.start() # 启动进程 p2.start() p1.join() # 等待子进程退出 p2.join() queue_obj.close() # 销毁队列
运行后的输出如下:
$ python3 queue1.py
queue读出进程开始运行 # 两个子进程启动了起来
queue写入进程开始运行
queue写入进程写入数据good # 写入good
queue读出进程读出数据good # 读出good
queue写入进程写入数据morning
queue读出进程读出数据morning
queue写入进程写入数据everyone # 写入everyone
queue读出进程读出数据everyone # 读出everyone
queue写入进程退出 # 子进程退出
queue读出进程退出
3) Lock锁
该对象主要提供了两个接口函数,一个是得到锁 acquire(),另外一个是释放锁 release()。下面的例子演示了 3
个进程,它们都往同一个文件写入数据。由于其每轮操作需要执行 3 次数据写入,而且要求在这 3 次操作之间其他的进程不能往文件写入数据。
可以使用锁来实现该功能,代码如下:
import multiprocessing import time def process_entry (lock, fd, id): # 进程入口函数 for x in range(30): # 30轮输出,每轮进行3次写操作 lock.acquire() # 得到锁 line = "%d: line 1, round: %d\n" % (id, x) time.sleep(0.1) fd.write(line) # 第1次写 fd.flush() line = "%d: line 2, round: %d\n" % (id, x) fd.write(line) # 第2次写 fd.flush() time.sleep(0.1) line = "%d: line 3, round: %d\n" % (id, x) fd.write(line) # 第3次写 fd.flush() lock.release() # 释放锁 time.sleep(0.1) if __name__ == "__main__": file_name = "shared_input2.txt" fd = open(file_name, "a+") lock = multiprocessing.Lock() # 创建锁 p1 = multiprocessing.Process(target=process_entry, args=(lock, fd, 1)) p2 = multiprocessing.Process(target=process_entry, args=(lock, fd, 2)) p3 = multiprocessing.Process(target=process_entry, args=(lock, fd, 3)) p1.start() # 启动进程 p2.start() p3.start() p1.join() # 等待子进程结束 p2.join() p3.join() fd.close() # 关闭文件 print(u"开始检查结果") # 开始检查结果 fd2 = open(file_name, "r") lines = fd2.readlines() fd2.close() line_num = len(lines) # 显示行数 print(u"总的行数: %d" % line_num) # 要求每连续的3行是同一个进程打印出来的 for x in range(int(line_num/3)): # 第1行和第2行不是同一个进程打印出来的 if lines[x*3][:2] != lines[x*3+1][:2]: print("line %d: Error" % x*3+1) # 发现错误,退出 sys.exit(1) # 第1行和第3行不是同一个进程打印出来的 if lines[x*3][:2] != lines[x*3+2][:2]: print("line %d: Error" % x*3+2) # 发现错误,退出 sys.exit(1) print(u"成功通过检查") # 所有行的检查通过
运行后的输出如下:
$ python lock2.py
开始检查结果
总的行数: 270
成功通过检查
也可以使用 with lock 语句,这样就不用再显式调用 acquire() 和 release() 了,只需将 acquire() 和 release() 之间的代码放到 with lock 块中即可。
下面的代码实现和前面一样的功能,但是使用 with lock 语句代替了 acquire() 和 release()。代码如下:
import multiprocessing import time def process_entry(lock, fd, id): for x in range(30): # 30轮操作,每轮3次写操作 with lock: # 每次写入3行,不能被打断 line = "%d: line 1, round: %d\n" % (id, x) time.sleep(0.1) fd.write(line) fd.flush() line = "%d: line 2, round: %d\n" % (id, x) fd.write(line) fd.flush() time.sleep(0.1) line = "%d: line 3, round: %d\n" % (id, x) fd.write(line) fd.flush() time.sleep(0.1) if __name__ == "__main__": file_name = "shared_input.txt" # 打开输出文件 fd = open(file_name, "a+") lock = multiprocessing.Lock() # 创建锁 p1 = multiprocessing.Process(target=process_entry, args=(lock, fd, 1)) p2 = multiprocessing.Process(target=process_entry, args=(lock, fd, 2)) p3 = multiprocessing.Process(target=process_entry, args=(lock, fd, 3)) p1.start() # 启动3个子进程 p2.start() p3.start() p1.join() # 等待子进程结束 p2.join() p3.join() fd.close() # 关闭文件 print(u"开始检查结果") # 检查结果 fd2 = open(file_name, "r") lines = fd2.readlines() fd2.close() line_num = len(lines) print(u"总的行数: %d" % line_num) for x in range(int(line_num/3)): # 如果第1行和第2行不是同一个进程打印 if lines[x*3][:2] != lines[x*3+1][:2]: print("line %d: Error" % x*3+1) # 发现错误,退出 sys.exit(1) # 如果第1行和第3行不是同一个进程打印 if lines[x*3][:2] != lines[x*3+2][:2]: print("line %d: Error" % x*3+2) # 发现错误,退出 sys.exit(1) print(u"成功通过检查") # 所有行检查通过
运行后的输出如下:
$ python lock1.py
开始检查结果
总的行数: 270
成功通过检查
通过运行结果可以发现使用 with lock 语句的效果和使用 acquire() 和 release() 的效果一样。
10
条内容
通常情况下,一个进程包括至少一个线程,如果有多个线程,其中包括一个主线程。同一个进程内的所有线程共享系统资源,但它们有各自独立的栈、寄存器环境和本地存储。
多线程的好处是可以同时执行多个任务,如果系统有多个计算单元,那么多个线程可以在各自的计算单元并行运行,这样可以极大提升系统的处理效率。
多数情况下进程比线程大,通常一个进程可以包含多个线程。进程的隔离效果比线程好,所以使用多进程会比使用多线程更加安全。多进程相对多线程的缺点是其调度比较重,效率比较低。