- 引子
- multiprocessing 模块
- multiprocessing.Process
- 1. Process 创建进程
- 2. Process 语法结构
- 3. 进程实现并发
- 4. join 方法
- 5. 进程间的隔离关系
- 6. 进程对象属性和方法
- 查看进程号
- 杀死子进程
- 判断进程是否存活
- 7. 僵尸进程与孤儿进程
- 8. 守护进程
- 9. 互斥锁
- 模拟简易抢票。
- 互斥锁介绍
- 改进抢票流程
- 进程间通信
- Queue 模块
- IPC机制
- 生产者消费者模型
- 基于队列实现生产者消费者模型
在前面的文章中知道了进程是正在运行的程序的实例，其实创建一个进程就是在内存中申请一个内存空间用于运行相应的程序代码。那么如何创建一个进程呢，有俩种方式，除了使用鼠标点击的方式来创建还可以通过代码来创建，而在python中需要用到 multiprocessing 模块
算起来 multiprocessing 不是一个模块而是python中一个 *** 作、管理进程的包。在这个包中几乎包含
了和进程有关的所有子模块。由于提供的子模块非常多，大致分为四个部分: 创建进程部分，进程同步部分，
进程池部分，进程之间数据共享。
Process 是 multiprocessing 的一个类，用于创建进程对象。通过创建一个 Process 对象然后调
用它的 start() 方法来生成进程。有以下俩种方式
1. 函数形式
代码示例
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
print('父进程')
输出结果
父进程
XWenXiang is running
XWenXiang finished his run
1. 我们在执行这个py文件的时候其实就相当于有了一个进程，称之为父进程
2. 示例中给类 Process 传入了函数以及函数的参数，得到一个进程对象，然后使用 start() 方法启动子
进程。传入的函数就是一个需要子进程中执行的任务。传参数的时候必须是一个'元组'，所以只有一个参数
的时候要在参数后加逗号表明这是元组。
3. 在'windows'中创建进程是以'导入模块'的方式从上往下依次读取父进程里的代码并放在子进程里面。并
执行前面传入的函数。在'linux和mac'中创建进程是'直接拷贝'一份源代码执行，不需要写在__main__中
4. 因为 Windows 是以导入模块的形式，我们就需要使用'__main__'进行一个限制，如果没有 __main__
在创建新进程时又读取到创建进程的语句，就会进入一个死循环。而 __main__ 可以限制只能在当前文件才
能创建进程。
5. 我们在输出结果中可以看到，“父进程”这三个字首先被打印出来，也就是说，父进程并没有等待子进程创
建完成，这种方式就是'异步'的形式
总结:
创建函数传入进程对象并以元组传参，windows 使用导入模块的形式读取父进程代码，并执行传入的
函数。父进程和子进程为异步形式。需要注意的是在 Windows 中别忘了 __main__ 方法。
2. 类形式
除了函数的方法，还可以通过继承 Process 类的形式创建进程
代码示例
from multiprocessing import Process
import time
class MyProcess(Process):
def __init__(self, username):
self.username = username
super().__init__()
def run(self):
print('早上好，', self.username)
time.sleep(3)
print('晚上好，', self.username)
if __name__ == '__main__':
p = MyProcess('XWenXiang')
p.start()
print('父进程')
输出结果
父进程
早上好， XWenXiang
晚上好， XWenXiang
1. 创建一个类并继承类Process。在类中定义'run()方法'，这是在进程启动时运行的方法，正是它去调用
target 指定的函数，我们自定义类的类中一定要实现该方法。传参的方式是使用 __init__方法，还需要
super 父类的方法
2. 同样的，Windows 中也需要使用 __main__ 来限定，不过此时创建进程对象是调用继承了类Process，
例如示例中的 MyProcess 类，然后传入参数即可，启动进程也是 start()。
Process([group [, target [, name [, args [, kwargs]]]]])
'target': 传递一个函数名，可以理解成子进程就是执行这个函数的代码，
'args': 给target指定的函数传递的参数，以元组的方式传递
'kwargs': 给target指定的函数传递关键字参数，以字典的方式传递
'name': 给进程设定一个名字，可以不设定
'group': 指定进程组，大多数情况下用不到
在前面的文章中只能实现一个客户端和服务端进行交互，我们还可以创建新进程来实现多个客户端和服务端进行交互。
4. join 方法
服务端代码服务端代码示例 import socket from multiprocessing import Process def run(sock, addr): while True: res = sock.recv(1024) print(f"{res.decode('utf8')} from {addr}") sock.send('我是服务端'.encode('utf8')) if __name__ == '__main__': s = socket.socket() s.bind(('127.0.0.1', 8080)) s.listen(5) while True: sock, addr = s.accept() p = Process(target=run, args=(sock, addr)) p.start() 1. 我们可以创建子进程，让子进程和客户端进行交互 2. 首先我们定义一个函数，函数里是子进程需要执行的语句，而我们只需要将通信功能语句放在函数里 面即可。其他的例如给服务端绑定地址代码 'bind()' 需要放在' __main__ '后面，这些代码我们不 需要子进程进行读取，要是子进程读取了'bind()'会造成路径冲突，会报以下错误: 'OSError: [WinError 10048] 通常每个套接字地址(协议/网络地址/端口)只允许使用一次。' 3.
客户端代码示例客户端代码示例 import socket c = socket.socket() c.connect(('127.0.0.1', 8080)) while True: info = input('>>> ').strip() c.send(info.encode('utf8')) res = c.recv(1024) print(res.decode('utf8'))
在上面写道父进程和子进程是异步方式进行的，也就是不需要等待子进程结束。若是需要父进程等待子进程运行结束，就需要用到 join 方法
代码示例
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
p.join()
print('父进程')
输出结果
XWenXiang is running
XWenXiang finished his run
父进程
1. 在一个已经被启动的子进程中使用 join() 方法，也就是说在 start() 后面使用 join() 方法，否则
会报 'AssertionError: can only join a started process' 错误，只能加入已启动的进程。
2. 使用了 join() 方法后，可以看见“父进程”在最后打印，这和前面创建进程的代码不同，前面的是异步方
式，而此时是同步的形式，父进程等子进程结束之后在进行自己打印 *** 作。
join 方法小练习
练习示例
from multiprocessing import Process
import time
def run(name, ):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p1 = Process(target=run, args=('XXX',)) # 创建一个进程对象
p2 = Process(target=run, args=('WWW',)) # 创建一个进程对象
first_time = time.time()
p.start()
p1.start()
p2.start()
p.join()
p1.join()
p2.join()
second_time = time.time()
print('父进程', second_time-first_time)
问，在练习中俩个时间的时间差大概是几秒？
答案是 3 秒左右。
(看起来好像是要等待 9 秒，但其实在第一个进程等待的时候，第二个第三个也在等待，同时进行的所以时间
会在 3 秒左右。)
若是将顺序换成这样就需要等待 9 秒了:
p.start()
p.join()
p1.start()
p1.join()
p2.start()
p2.join()
进程之间默认隔离，也就是说一个进程修改了同名的变量，另外一个进程不会受影响。因为是默认隔离，所以其实也有修改的方式。
代码示例
from multiprocessing import Process
money = 999
def task():
global money # 局部修改全局不可变类型
money = 666
if __name__ == '__main__':
p = Process(target=task)
p.start()
p.join() # 确保子进程代码运行结束再打印money
print(money)
输出结果
999
1. 首先在父进程中定义了一个 money 变量，定义一个函数，函数体里修改全局变量 money
2. 在子进程中执行该函数，并让父进程等待子进程结束。
3. 此时打印的父进程变量值没有变化，也就是在子进程中修改 *** 作不会影响父进程。
进程号简单说就是 *** 作系统里用于唯一标识进程的一个数值。我们获取进程号的用处之一就是可以通过代码的方式管理进程
cmd 查看
windows: tasklist结果集中PID
mac: ps -ef
current_process函数
在 multiprocessing 模块中定义了一个函数 current_process， 调用这个函数可以获得进程号(结果不
固定)
代码示例
from multiprocessing import Process, current_process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('子进程的进程号: ', current_process().pid) # 获取子进程号
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
print('父进程的进程号: ', current_process().pid) # 获取父进程号
os 模块
使用 os 模块也可以，其有俩个方法:
os.getpid() # 获取当前进程的进程号
os.getppid() # 获取当前进程的父进程号
代码示例
from multiprocessing import Process
import time
import os
def run(name):
print('%s is running' % name)
time.sleep(10)
print('子进程的进程号: ', os.getpid())
print('父进程的进程号: ', os.getppid())
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
print('父进程的进程号: ', os.getpid())
print('父进程的父进程进程号: ', os.getppid())
1. 父进程的父进程其实是 pycharm 。
杀死进程其实就是将进程从内存中清理掉结束掉。
cmd 杀死进程
windows taskkill关键字
mac/linux kill关键字
代码实现使用的是terminate()方法
代码示例
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
p.terminate() # 杀死该进程对象
print('父进程')
输出结果
父进程
1. 由于杀死了该进程，所以该进程并不会执行。
is_alive()可以判断该进程是否在执行，若是在执行就返回一个 True。
代码示例
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
p.terminate() # 杀死进程
res = p.is_alive()
print(res)
输出结果
True
1. 奇怪，为什么明明杀死了这个进程但是判断出来还是 True 呢，因为此时进程是异步 *** 作，然而杀死进程
需要一定的时间，也就是说判断的比杀死的要早。
代码示例
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.start() # 告诉 *** 作系统创建一个新的进程
p.terminate()
p.join()
res = p.is_alive()
print(res)
输出结果
False
1. 在第一个示例的基础上在加了 join() 方法，让父进程等到子进程结束在运行判断子进程是否存活。
（如果此时 join() 方法放在 terminate() 方法前面，会先执行子进程在执行杀死进程语句）
'僵尸进程':
子进程是通过父进程创建的，且两者的运行是相互独立的，父进程永远无法预测子进程什么时候结束
当子进程调用 exit 命令结束自己的生命时，其实它并没有真正的被销毁，内核只是释放了该进程的所有资源
(包括打开的文件、占用的内存等)，但是留下一个称为僵尸进程的数据结构，这个结构保留了一定的信息（包
括进程号pid ，退出状态，运行时间），这些信息直到父进程通过 wait()/waitpid() 来取时才释放。
'孤儿进程':
顾名思义。一个父进程退出，而它的一个或多个子进程还在运行，那么这些子进程将成为孤儿进程。
什么是守护进程，守护进程就是会随着主进程结束而结束的进程。可以想象成一个骑士守护着一个公主，公主要是挂了骑士会追随她而去哈哈。
守护进程有俩个特点:
1. 守护进程会在主进程代码执行结束后就终止
2. 守护进程内无法再开启子进程,否则抛出异常：
'AssertionError: daemonic processes are not allowed to have children'
创建守护进程是将 daemon 改为 True
代码示例:
from multiprocessing import Process
import time
def run(name):
print('%s is running' % name)
time.sleep(3)
print('%s finished his run' % name)
if __name__ == '__main__':
p = Process(target=run, args=('XWenXiang',)) # 创建一个进程对象
p.daemon = True # 将进程设置成守护进程
p.start() # 告诉 *** 作系统创建一个新的进程
print('父进程')
输出结果
父进程
'''
需要注意的是，给一个进程设置成守护进程的的时候要放在 start() 启动前，否则会抛出异常:
AssertionError: process has already started
'''
1. 在示例中，异步的情况，父进程已经结束，而守护进程还没来得及执行就被挂了。
在了解互斥锁之前我们先看一个例子，我们是不是经常在网上抢票或者购物的时候，显示还有库存，但是当我
们点击购买的时候却提示库存不足。这是因为页面没刷新，假设我们看的界面是8点钟的时候，由于某些原因到
了9点才继续看手机，但是此时的界面是8点钟的界面，数据不会改变。
fare.json 文件（存放票数）fare.json 文件代码示例: {"fare_num": 1}
index.py 文件（用于产生子进程）index.py 文件代码示例: from multiprocessing import Process import json import time import random file_path = 'fare.json' def check(name): with open(file_path, 'r', encoding='utf8') as f: res = json.load(f).get('fare_num') print(f"用户 {name} 查询余票 {res} 张") def buy(name): with open(file_path, 'r', encoding='utf8') as f: res = json.load(f) time.sleep(random.randint(1, 3)) # 随机等待 if res['fare_num'] > 0: res['fare_num'] -= 1 with open(file_path, 'w', encoding='utf8') as e: json.dump(res, e) print(f'用户 {name} 抢票成功') else: print(f'用户 {name} 抢票失败,库存不足') def run(name): check(name) buy(name) if __name__ == '__main__': for i in range(1, 5): p = Process(target=run, args=(i,)) p.start() 1. 买票首先是查看票数，然后买票的时候需要在看一次票数。票数小于 0 则购票失败 2. 使用 for 循环产生指定数量的子进程，也就是用户。由于 for 循环瞬间的告诉 *** 纵系统需要创建 多个子进程，具体的谁先谁后得看 *** 作系统。所以可以看成是同时产生多个子进程。 3. 但是此时运行的结果是全部用户都可以抢到唯一的票，是因为当多个进程 *** 作同一份数据的时候会造 成数据的错乱，并没有达到预期结果。
当多个进程对一个数据进行 *** 作的时候，很容易错乱，所以需要给数据上锁。加锁可以保证多个进程修改同一个数据时，同一时间只能有一个进程可以进行修改，即将并行改成串行。虽然速度减慢了但是安全性得到很大的提高。
互斥锁只在处理数据的部分加锁，可以想象成，一个房间一次只能进入一个人，进入了之后将房间锁起来，等
到第一个人 *** 作结束打开锁出门下一个人才能进去。
在代码中，需要引入 Lock ，其主要有以下几个 *** 作:
mutex = Lock() # 相当于产生一把钥匙
mutex.acquire() # 抢锁
mutex.release() # 放锁
锁相关知识
行锁:针对行数据加锁 同一时间只能一个人 *** 作
表锁:针对表数据加锁 同一时间只能一个人 *** 作
悲观锁:就是以悲观的态度处理，就是在修改数据之前就把数据加锁，直到前面的将锁释放才能修改
乐观锁:对数据冲突保持乐观的态度，不对数据进行加锁，只有在提交数据的时候会去验证数据是否
冲突(例如加版本号)，这样的方式由于都可以提交请求，但多数无用功，所以不适合高并发
index.py 文件（用于产生子进程）
改进代码
from multiprocessing import Process, Lock
import json
import time
import random
file_path = 'fare.json'
def check(name):
with open(file_path, 'r', encoding='utf8') as f:
res = json.load(f).get('fare_num')
print(f"用户 {name} 查询余票 {res} 张")
def buy(name):
with open(file_path, 'r', encoding='utf8') as f:
res = json.load(f)
time.sleep(random.randint(1, 3))
if res['fare_num'] > 0:
res['fare_num'] -= 1
with open(file_path, 'w', encoding='utf8') as e:
json.dump(res, e)
print(f'用户 {name} 抢票成功')
else:
print(f'用户 {name} 抢票失败,库存不足')
def run(name, mutex):
check(name)
mutex.acquire() # 抢锁
buy(name)
mutex.release() # 放锁
if __name__ == '__main__':
mutex = Lock()
for i in range(1, 5):
p = Process(target=run, args=(i, mutex))
p.start()
1. 在原有基础上给修改数据的部分加上一把锁，使其排队修改，而查看数据还是同时查看没有改变。
2. 这样就可以满足预期效果
前面说进程之间是互相不影响的，那么如果需要发送数据我们可以使用队列的方式。
需要知道队列是先进先出的，而堆栈是先进后出的。在python中可以使用内置队列来实现进程间通信。
在模块 multiprocessing 中除了 Queue ，还有 SimpleQueue 以及 JoinableQueue 队列类型。在这里介绍 Queue
在 queue 模块中有这么几个比较常用的方法:
q = Queue(5) # 生成一个队列对象，括号里的数字是队列的长度
q.put() # 将一个item放入到队列中
q.get() # 将一个item从队列中取出
--- q.full() # 判断队列是否已满，如果队列是满的返回 True ，否则返回 False
|-- q.empty() # 判断队列是否为空，如果队列为空，返回 True ，否则返回 False
|-- q.get_nowait() # 队列中如果没有值，则直接报错
|-- q.qsize() # 获得队列的大致长度
|
|______________ 在并发的时候不能精确的使用
代码示例
from multiprocessing import Queue
q = Queue(3) # 创建一个队列对象，可容纳最大长度为3
q.put(11) # 往队列中传值
q.put(22)
q.put(33)
print(q.full()) # 判断队列是否为已满，此时已满
print(q.empty()) # 判断队列是否为空，此时有值，不为空
print(q.qsize()) # 获取此时队列的大致长度
# q.put(44) 此时队列已满，如果继续传值，程序会暂停，等待队列中的数值被取出留出空位
print(q.get()) # 从队列中取值
print(q.get())
print(q.get())
# print(q.get()) 此时队列的item已被取完，继续取程序会等待，直到队列新增数据
# q.get_nowait() 此时队列没有值，运行此代码会报错
打印结果
True
True
3
11
22
33
简单介绍了 Queue 模块的使用后，我们可以使其和进程结合起来，在进程和进程之间形成一个共享的进程队列，也就是说使用了 Queue 可以实现多进程之间的数据传递。这就是IPC的概念
代码示例
from multiprocessing import Queue, Process
def get_item(q): # 定义获取队列值的函数
res = q.get()
print(f"{res} 真帅")
def put_item(q): # 定义给队列传递值的函数
q.put('XWenXiang')
if __name__ == '__main__':
q = Queue(5) # 定义一个队列
p = Process(target=get_item, args=(q,)) # 定义子进程用于取数据
p1 = Process(target=put_item, args=(q,)) # 定义子进程用于传数据
p.start()
p1.start()
print('主进程')
输出结果
主进程
XWenXiang 真帅
此时，俩个子进程就可以做到数据交互，当然主进程和子进程之间同样可以。
'生产者消费者模型'
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费
线程的工作能力来提高程序的整体处理数据的速度。
'为什么要使用生产者和消费者模式'
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果
生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据
同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是
引入了生产者和消费者模式。
'什么是生产者消费者模式'
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不
直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞
队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产
者和消费者的处理能力。
我们可以简单的模拟生产者消费者模型，其实和前面的示例相似，只不过引进了一个名称概念。
代码示例
from multiprocessing import Process, Queue
import time
import random
def make(name, q):
for i in range(1, 6):
time.sleep(random.random())
q.put('包子')
print(f"{name} 做了第 {i} 个包子")
def eat(name, q):
while True:
time.sleep(random.random())
q.get()
print(f"{name} 吃了包子")
if __name__ == '__main__':
q = Queue()
p = Process(target=make, args=('小王', q))
p1 = Process(target=eat, args=('老王', q))
p.start()
p1.start()
"""队列中其实已经加了锁，所以多进程取值也不会冲突，并且取走了就没了"""
1. 定义了俩个子进程，分饰生产者和消费者。示例中举的是制作包子和吃包子且加上了时间暂停，很明显可
以看到生产者只向队列传了5个数据，也就是说只生产了5个包子，但是消费者在不停的吃会导致程序暂停，等
待生产者生成数据。
上面的案例如果消费者过多程序不会停止，我们可以浅用一下 JoinableQueue 模块
代码示例
from multiprocessing import Process, JoinableQueue
import time
import random
def make(name, q):
for i in range(1, 6):
time.sleep(random.random())
q.put('包子')
print(f"{name} 做了第 {i} 个包子")
def eat(name, q):
while True:
time.sleep(random.random())
q.get()
q.task_done() # 每次取完数据必须给队列一个反馈
print(f"{name} 吃了包子")
if __name__ == '__main__':
q = JoinableQueue()
p = Process(target=make, args=('小王', q))
p1 = Process(target=eat, args=('老王', q))
p1.daemon = True # 将消费者设定成守护进程，随主进程结束而结束
p.start()
p1.start()
p.join() # 等待生产者停止生产，队列数据不在增加
q.join() # 等待队列中数据全部被取出
此时没有继续生产程序会结束。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）