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() 的效果一样。
湘公网安备 43102202000103号