python中多进程

Posted 2020-11-19 xiaoyafei

多进程

什么是进程

进程：正在进行的一个过程或者说一个任务，而负责执行任务的是CPU。

进程和程序的区别

程序仅仅是一堆代码而已，而进程指的是程序的运行过程。

举例

想象以为有着一手好厨艺的科学家肖亚飞正在为自己的女儿烘焙蛋糕，他有着做生日蛋糕的食谱，厨房里有所需要的原料：面粉、鸡蛋、韭菜、蒜泥等。

在这个比喻中

做蛋糕的食谱就是程序（即用适当形式描述的算法）

计算机科学家就是处理器（CPU）

而做蛋糕的各种原料就是输入数据

进程就是厨师阅读食谱、取来各种原料以及烘焙蛋糕等一系列动作的总和

现在假设科学家的儿子哭着跑了进来，说：我的头被蜇伤了

科学家想了想，处理儿子蜇伤的任务比给女儿烘焙蛋糕的任务更重要，于是

科学家就记录下了他照着食谱做到哪儿了（保存进程的当前状态），然后拿出一本急救手册，按照其中的指示处理蜇伤。这里，我们看到处理机从一个进程（做蛋糕）切换到另一个高优先级（实施医疗救治），每个进程拥有各自的程序（食谱和急救手册）。当蜇伤处理完之后，这位科学家又回来做蛋糕，从他离开时的那一步继续做下去。

需要强调的是：同一个程序执行两次，那也是两个进程，比如打开暴风影音，虽然都是一个软件，但是一个可以播放走火，一个可以播放爱国者。

并发和并行

无论是并发还是并行，在用户看来都是‘同时’运行的，不管是进程还是线程，都只是一个任务而已，真实干活的是CPU，CPU做这些任务，而一个CPU同一时刻只能执行一个任务。

并发：是伪并行，即看起来是同时运行。单个CPU+多道技术就可以实现并发

举例：（单核+多道，实现多个进程的并发执行）

我在一个时间段内有很多事情要做：王者荣耀上分、看《爱国者》、交女朋友，但是我在同一时刻只能做一个任务（CPU在同一时间只能干一个活），如何才能玩出多个任务并发执行的效果呢？

就是我先打两局王者荣耀，然后看一会电视剧，再去和女朋友聊聊天......这样就保证了每个任务都在进行中。

并行：同时运行，只有具备多个CPU才能实现并行

单核下，可以利用多道技术，多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）

有4个核，有6个任务，这样同一时间有4个任务就被分配了，假设分别分配给了cpu1、cpu2、cpu3、cpu4，一旦在任务1中遇到I/O堵塞就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术。

而一旦任务1的I/O结束了，操作系统会重新调用它（需要知道进程的调度、分配给哪个cpu运行，由操作系统说了算），可能会分配给4个cpu的任意一个去执行

进程的创建

但凡是硬件，都需要有操作系统去管理，只要有操作系统，就有进程的概念，就需要有创建进程的方式，一些操作系统只为了一个应用程序设计，比如微波炉中的控制器，一旦启动微波炉，所有的进程都已经存在。

而对于通用系统（跑很多应用程序），需要有系统运行过程中创建或者撤销进程的能力，主要分为4种形式创建新的进程：

• 1.系统初始化（查看进程linux中用ps命令，windows中要用任务管理器，前台进程负责与用户交互，后台运行的程序与用户无关，运行在后台并且只在需要时被唤醒的进程，称为守护进程，如：电子邮件、web界面、打印）
• 2.一个进程在运行过程中开启了子进程（如：nginx开启多进程，os.fork，subprocess.Popen等）
• 3.用户的交互式请求，而创建了一个新的进程（如：用户双击暴风影音）
• 4.一个批处理作业的初始化（只在大型机的批处理系统中应用）

无论哪一种，新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建的：

1.在UNIX系统中，该系统调用的是fork，fork会创建一个与父进程一模一样的副本，二者拥有相同的存储映像、同样的环境字符串和同样的打开文件（在shell解释器进程中，执行一个命令就会创建一个子进程）

2.在windows中该系统调用的是CreateProcess，CreateProcess既处理进程的创建，也负责把正确的程序装入新进程

关于创建子进程，unix和windows的相同点和不同点

相同点：进程创建后，父进程和子进程有各自不同的地址空间（多道技术要求物理层面实现进程之间内存的隔离），任何一个进程的在其地址空间中的修改都不会影响到另外一个进程

不同点：在UNIX中，子进程的初始地址空间是父进程的一个副本，提示：子进程和父进程是可以有只读的共享内存区的。但是在windows中，从一开始父进程和子进程的地址空间就是不同的。

进程的状态

运行态：应用程序正在被CPU执行中

阻塞态：当前进程突然要做I/O操作，然后CPU去执行其他的程序

就绪态：时刻准备着能够被执行

好了，讲了这么多理论知识，现在就让我们看看如何开启进程的吧？

开启进程的两种方式

# 开启进程的第一种方式

from multiprocessing import Process
import time

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:  # windows下开启进程的指令需要放在main下面
    p = Process(target=task,args=(‘子进程1‘,))  # target代表去执行一个任务，如果是加括号的话相当于立马就执行了
    p.start()
    print(‘主进程‘)
    
# 运行结果如下：
主进程
子进程1 is running
子进程1 is done

# 开启进程的第二种方式
from multiprocessing import Process
import time

class MyProcess(Process):  # 定制自己的方法
    def __init__(self,name):
        super(MyProcess, self).__init__()  # 重写父类方法
        self.name = name

    def run(self):  # 这里一定要用run
        print(‘%s is running‘%self.name)
        time.sleep(2)
        print(‘%s is done‘%self.name)

if __name__ == ‘__main__‘:
    # 实例化4个对象
    p1 = MyProcess(‘子进程1‘)
    p2 = MyProcess(‘子进程2‘)

    p1.start()  # 会自动调用run方法
    p2.start()
    print(‘主进程‘)  # 首先第一步肯定是先打印出这句话
    
# 运行结果为：
主进程
子进程2 is running
子进程1 is running
子进程2 is done
子进程1 is done

我们看到子进程2先运行了，那么程序应该是从上到下执行，先执行子进程1然后再执行子进程2才对呀，这个我们是不可以控制谁先执行谁后执行的，因为启动的速度太快了，等到后面我可以教你如何先启动子进程1再启动子进程2的

刚在上面说到，创建子进程的时候，会把父进程/主进程的数据复制一份作为子进程的初始数据，但进程之间的数据是共享的还是隔离的呢？让我们来证明一下：

from multiprocessing import Process
import time

num = 100  # 定义一个全局变量，是属于主进程的
def task():
    global num
    num = 10
    print(‘子进程中n的值为：‘,num)

if __name__ == ‘__main__‘:
    p1 = Process(target=task)
    p1.start()
    print(‘主进程中N的值为：‘,num)
    
# 运行结果如下：
主进程中N的值为： 100
子进程中n的值为： 10

那么经过这一段代码就可以证明，进程之间的数据是不共享的

现在让我们来尝试一下：基于多进程并发实现的套接字通信，但是首先我会将套接字通信给先写出来，就算给大家复习了：

#!/usr/bin/python3

# socket客户端
import socket
server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘,8080))
server.listen()
print(‘starting......‘)

conn,addr = server.accept()
print(addr)

while True:
    try:
        data = conn.recv(1024)
        print(‘客户端的数据：‘,data)
        conn.send(data.upper())
    except ConnectionResetError:
        break
conn.close()
server.close()

#!/usr/bin/python3

# socket客户端
import socket
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect((‘127.0.0.1‘,8080))

while True:
    msg = input(‘>>>‘).strip()
    if not msg:continue
    client.send(msg.encode(‘utf-8‘))
    data = client.recv(1024)
    print(data.decode(‘utf-8‘))

client.close()

在 以上这段代码中，服务端开放8080端口用于socket通信，客户端连接本地8080端口实现通信，但是这段代码中有一个很明显的缺陷就是：服务端一次只能和一个客户端进行连接，试换到web服务器上，一个客户端1过来连接了，那么下一个客户端2一直要等到这个客户端1连接分手之后才可以进行连接吗？答案肯定不是这样的，我们之前说过，一个程序运行两次就是两个进程，所以这个代码我们需要使用多进程改写：

#!/usr/bin/python3

# socket客户端
import socket
from multiprocessing import Process

# 通信任务
def talk(conn):
    while True:
        try:
            data = conn.recv(1024)
            print(‘客户端的数据：‘,data)
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()

# 连接任务
def server(ip,port):
    server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
    server.bind((ip,port))
    server.listen()

    while True:
        conn,addr = server.accept()
        p = Process(target=talk,args=(conn,))
        p.start()

    server.close()

if __name__ == ‘__main__‘:
    server(‘127.0.0.1‘,9991)
    
    
#!/usr/bin/python3

#socket客户端
import socket
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect((‘127.0.0.1‘,9991))

while True:
    msg = input(‘>>>‘).strip()
    if not msg:continue
    client.send(msg.encode(‘utf-8‘))
    data = client.recv(1024)
    print(data.decode(‘utf-8‘))

client.close()

那么代码经过这样的改动的话，服务端开放9991端口后，每运行一个客户端就相当于一个进程去连接服务端，就实现了基于socket网络通信+多进程的方式了

查看进程的PID

在操作系统中，有一个进程ID号，代表着当前这个进程运行着什么功能。

# 在linux中使用ps aux就可以查看，其中‘T‘代表停止状态，表示目前PID暂时没有被回收
root       1197  0.0  0.6 131932  6664 pts/2    T    17:15   0:00 python3 客户端.py
root       1198  0.0  0.6 141588  6416 pts/0    T    17:15   0:00 python3 服务端.py

让我们来看下子进程的PID，在windosw下举例：

from multiprocessing import Process
import os
import time

def task():
    print(‘%s is running,parent id is <%s>.‘%(os.getpid(),os.getppid()))  # 子进程ID，父进程ID
    time.sleep(3)
    print(‘%s is done,parent id is <%s>.‘%(os.getpid(),os.getppid()))

if __name__ == ‘__main__‘:
    p = Process(target=task,)
    p.start()
    print(‘主进程‘,os.getpid(),os.getppid())  # 主进程的ID，主进程的父亲ID
    
# 那么运行结果是这样的
主进程 5308 8332
10408 is running,parent id is <5308>.
10408 is done,parent id is <5308>.

我们之前说过了，当一个主进程中开启子进程，那么子进程会去拷贝父进程中的数据作为子进程的原始数据，so，主进程的ID是10408，父进程的ID是5308。

那么8332是个什么鬼，我给你看个东西你就懂了

C:Usersxiaoyafei>tasklist | findstr pycharm
pycharm64.exe                 8332 Console                    3  1,108,664 K

看到了吗？是pycharm的进程号。

僵尸进程和孤儿进程(了解)

所谓僵尸进程：就是在主进程开启了一个子进程后，无论什么时候都可以去查看子进程的状态，即使子进程死掉了，也要为主进程保留子进程状态信息，僵尸进程是有害的，因为一个进程死掉后，它的PID不会立马消除，如果僵尸进程多了，PID还被占用着，如果操作系统再开启新的进程的话可能就起不来，在父进程一直不死的情况下是有害的。

所谓孤儿进程：就是子进程还没有执行完，主进程就已经死掉的了，但是子进程是无害的，此时子进程的PID由init进程去回收。

Process对象的其他属性和方法

join()方法，主进程等到子进程完成之后再去执行

在主进程运行过程中，如果想要并发的执行任务，我们可以开启子进程，此时主进程的任务和子进程的任务分两种情况：

1.在主进程的任务与子进程的任务彼此独立的情况下，主进程的任务先执行完毕后，主进程还需要等待子进程执行完毕，然后统一回收资源。

2.如果主进程的任务在执行到某一阶段后，需要等待子进程执行完毕后才能继续执行，就需要有一种机制能够让主进程检测子进程进程是否运行完毕，在子进程执行完毕后才继续执行，否则一直在原地阻塞，这就是join方法的作用。

from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s is running.‘%self.name)
        time.sleep(2)
        print(‘%s is done.‘%self.name)

if __name__ == ‘__main__‘:  # windows系统需要在main下开启进程
    p = MyProcess(‘子进程‘)  # 实例化子进程
    p.start()  # 给操作系统发送信号，把父进程的数据拷贝给子进程作为初始数据
    p.join()  # join方法，等到子进程完成之后，才会执行主进程
    print(‘主进程‘)

# 运行结果如下：
子进程 is running.
子进程 is done.
主进程

有的人可能会问了，有了join()方法的话，程序不就变成串行了吗？在这里解释一下：

进程只要start就会在开始运行了，所以p.start()时，系统中已经有了1个并发的进程了，而我们p.join()是在等p结束，没错p只要不结束主线程就不会执行，这也是问题的关键，join是让主线程在等，而p或者以后的p1/p2/p3仍然是并发执行的，等到p.join()结束，可能p1/p2/p3早都结束了，这样的话，p1/p2/p3就忽略了检测，无需等待。

所以不管有多少个join()方法，需要等到的时候仍然是耗费时间最长的那个进程运行的时间。

terminate和is_alive方法

其中is_alive()是检测进程是否存活，而terminate方法是用来关闭进程的，当然不会立马关闭

# is_alive()检测进程是否存活
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self,name):
        super(MyProcess, self).__init__()
        self.name = name

    def run(self):
        print(‘%s is running‘%self.name)
        time.sleep(3)
        print(‘%s is done‘%self.name)

if __name__ == ‘__main__‘:
    p1 = MyProcess(‘子进程1‘)
    p1.start()
    print(‘子进程进程是否存活：‘,p1.is_alive())
    p1.join()
    print(‘主进程‘)
    print(‘第二次检测是否存活：‘,p1.is_alive())
    
# 运行结果为
子进程进程是否存活： True
子进程1 is running
子进程1 is done
主进程
第二次检测是否存活： False

这是由于主进程完成了任务，所以子进程就跟着主进程一起死掉了。

# terminate杀死进程，不会立马杀死掉
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self,name):
        super(MyProcess, self).__init__()
        self.name = name

    def run(self):
        print(‘%s is running‘%self.name)
        time.sleep(3)
        print(‘%s is done‘%self.name)

if __name__ == ‘__main__‘:
    p1 = MyProcess(‘子进程1‘)
    p1.start()
    p1.terminate()  # 杀死子进程1
    print(‘第一次检测子进程进程是否存活：‘,p1.is_alive())
    p1.join()
    print(‘主进程‘)
    print(‘第二次检测是否存活：‘,p1.is_alive())

运行结果为：
第一次检测子进程进程是否存活： True
主进程
第二次检测是否存活： False

在p1子进程刚刚把信号传递给操作系统之后，就利用了terminate方法杀死了子进程1，但是不是立马杀死，所以此时的p1子进程还是属于存活状态，等到打印完‘主进程‘之后，p1就跟随着主进程死掉了，所以此时子进程存活状态为False。

name和pid方法

其中，查看当前进程PID为os.getpid()方法，os.getppid()为查看父进程的PID方法

pid方法刚刚已经说过了，所以就不举例了

# name方法
from multiprocessing import Process
import time

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:
    p1 = Process(target=task,args=(‘xiao‘,))
    p1.start()
    p1.join()
    print(p1.name)  # 打印子进程的进程名
    print(‘主进程‘)

# 运行结果为：
xiao is running
xiao is done
Process-1  # 子进程的进程名，默认的，可以修改
主进程

# 如果想要修改进程名，只需要在实例化的时候添加name属性就可以了，具体操作为：
    p1 = Process(target=task,args=(‘xiao‘,),name=‘子进程1‘)

守护进程

主进程设置进程，然后将该进程设置成自己的守护进程。

关于守护进程需要强调两点：

1.守护进程会跟随者主进程代码执行结束后终止

2.守护进程内无法再开启子进程，否则会抛出异常

如果我们有两个任务需要并发执行，那么我们需要开一个主进程和一个子进程去执行，如果子进程的任务在主进程执行完成后没有存在的必要了，那么该子进程应该在开启前就被设置成守护进程。主进程代码运行结束，守护进程随之终止。

# 守护进程daemon
from multiprocessing import Process
import time

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:
    p = Process(target=task,args=(‘子进程‘,))
    p.daemon = True  # 设置为守护进程
    p.start()
    print(‘主进程......‘)

验证守护进程内无法再开启子进程

from multiprocessing import Process
import time

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

    p1 = Process(target=time.sleep,args=(3,))
    p1.start()


if __name__ == ‘__main__‘:
    p = Process(target=task,args=(‘子进程‘,))
    p.daemon = True
    p.start()
    p.join()  # 因为是验证无法再开启子进程，所以需要在添加前能保证子进程的程序能运行完
    print(‘主进程......‘)

# 那么运行结果是：反正就是一堆的报错
子进程 is running
Process Process-1:
子进程 is done
Traceback (most recent call last):
  File "D:pythonpython3libmultiprocessingprocess.py", line 258, in _bootstrap
    self.run()
  File "D:pythonpython3libmultiprocessingprocess.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "D:py_studyday22-多线程开始	est.py", line 16, in task
    p1.start()
  File "D:pythonpython3libmultiprocessingprocess.py", line 103, in start
    ‘daemonic processes are not allowed to have children‘
AssertionError: daemonic processes are not allowed to have children
主进程......

互斥锁

进程之间的数据是不共享的，但是共享同一套文件系统，所以访问同一个文件，或者同一个打印终端，是没有问题的，而共享带来的就是竞争，竞争带来的就是错乱，如下：

# 并发执行，效率高，但竞争着同一个打印终端，所以会带来错乱
from multiprocessing import Process
import time,os

def task():
    print(‘%s is running‘%os.getpid())
    time.sleep(2)
    print(‘%s is done.‘%os.getpid())

if __name__ == ‘__main__‘:
    for i in range(5):
        p = Process(target=task)
        p.start()

# 运行结果为：
13580 is running
7912 is running
6176 is running
13668 is running
12172 is running
7912 is done.
13580 is done.
6176 is done.
13668 is done.
12172 is done.

如何控制，就是加锁处理。而互斥锁的意思就是相互排斥，如果把多个进程比喻成多个人，那么互斥锁的工作原理就是多个人都要去争抢同一个资源：洗手间，一个人抢到了洗手间的锁，其余的人就要都等着，等到这个任务执行完成后释放锁，其他人中的一个人才有可能抢到这把锁......所以互斥锁的原理就是：把并行改为串行，降低了效率，但是保证了数据的安全不错乱。

from multiprocessing import Process,Lock
import time,os

def task(Lock):
    Lock.acquire()  # 加锁
    print(‘%s is running‘%os.getpid())
    time.sleep(2)
    print(‘%s is done.‘%os.getpid())
    Lock.release()  # 释放锁

if __name__ == ‘__main__‘:
    lock = Lock()  # 先实例化一个对象
    for i in range(5):
        p = Process(target=task,args=(lock,))
        p.start()
# 运行结果为
13868 is running  # 一个子进程开始了
13868 is done.  # 这个子进程死掉了
9576 is running
9576 is done.
13956 is running
13956 is done.
11696 is running
11696 is done.
11632 is running
11632 is done.

我们接下来举一个抢票的例子大家可能就懂了

模拟抢票

我们在12306上抢票过程中，明明看到了仅剩下1张票，但是现在呢有10个人在开始抢票，让我们来模拟一下：

from multiprocessing import Process
import json
import time

# 查询
def search(name):
    time.sleep(1)
    dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))  # 在当前目录下db.txt内容：
    print(‘<%s> 查看到剩余票数【%s】‘ %(name,dic[‘count‘]))

# 购票
def get(name):
    time.sleep(1)
    dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
    if dic[‘count‘] > 0:
        dic[‘count‘]-= 1
        time.sleep(3)
        json.dump(dic,open(‘db.txt‘,‘w‘,encoding=‘utf-8‘))
        print(‘<%s> 购票成功‘ %name)


def task(name):
    search(name)
    get(name)

if __name__ == ‘__main__‘:
    for i in range(10):
        p=Process(target=task,args=(‘路人%s‘ %i,))
        p.start()
# 运行结果
<路人2> 查看到剩余票数【1】
<路人0> 查看到剩余票数【1】
<路人3> 查看到剩余票数【1】
<路人4> 查看到剩余票数【1】
<路人1> 查看到剩余票数【1】
<路人8> 查看到剩余票数【1】
<路人5> 查看到剩余票数【1】
<路人9> 查看到剩余票数【1】
<路人6> 查看到剩余票数【1】
<路人7> 查看到剩余票数【1】
<路人2> 购票成功
<路人0> 购票成功
<路人3> 购票成功
<路人4> 购票成功
<路人1> 购票成功
<路人8> 购票成功
<路人5> 购票成功
<路人9> 购票成功
<路人6> 购票成功
<路人7> 购票成功

这样看运行结果的话，肯定是不合理的，票只有一张，怎么会让10个人都购票成功呢？所以这里需要使用互斥锁，互斥锁就是相互排斥，它工作的原理就是把并发变成串行，虽然程序运行效率低了，但是对数据安全和不错乱得到了明显的提升。

from multiprocessing import Process,Lock
import json
import time

def search(name):
    time.sleep(1)
    dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
    print(‘<%s> 查看到剩余票数【%s】‘ %(name,dic[‘count‘]))


def get(name):
    time.sleep(1)
    dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
    if dic[‘count‘] > 0:
        dic[‘count‘]-= 1
        time.sleep(3)
        json.dump(dic,open(‘db.txt‘,‘w‘,encoding=‘utf-8‘))
        print(‘<%s> 购票成功‘ %name)


def task(name,lock):
    ‘‘‘
    因为在查询的时候看到的余票是一样的，所以需要在购票环节需要添加互斥锁
    :param name:
    :param lock:
    :return:
    ‘‘‘
    search(name)
    lock.acquire()  # 加锁
    get(name)
    lock.release()  # 解锁

if __name__ == ‘__main__‘:
    # 首先要生成互斥锁对象
    lock = Lock()
    for i in range(10):
        p=Process(target=task,args=(‘路人%s‘ %i,lock))
        p.start()

# 运行结果为

在上面的代码中，10位用户都看到了剩余的票数，但是只有路人3抢到了票，这就相当于10个人去抢洗手间，但是只有3号用户得到了洗手间的钥匙进去了，并且把洗手间给反锁了。

互斥锁和join的区别

在之前呢，我们讲过了join()方法和互斥锁的概念，让我们来说一下他们之间的区别

join()方法的作用是让主进程等待子进程运行成功之后在再去运行，join()是把并行改成了串行，确实能够保证数据的安全以及不错乱，但是查票的过程中谁先查到票，那么这张票就是谁的。join()就是把所有的任务都变成了串行，如task函数里的search和get。可以比喻成在一段代码中在最上面添加了try，在最下面添加了except。

而互斥锁呢就是相互排斥，在购票环节设置互斥锁，也是将并行改成了串行，但是互斥锁是将程序中的一个任务的某一段代码设置成串行，就比如task函数里的get任务，然后互斥锁却十分符合我们的需求。

互斥锁总结

加锁可以保证多个进程在修改同一块数据的时，同一个时间只能有一个任务可以进行修改，即串行的修改，虽然效率下来了，但是却可以保证数据的安全性。

虽然可以用文件共享数据实现进程间通信，但问题是：

• 效率低（共享数据基于文件，而文件是硬盘上的数据）
• 需要自己加锁处理

因此我们需要找一种解决办法能够兼顾：

• 效率高（多个进程共享一块内存的数据）
• 帮我们处理好锁的问题

这就是mutliprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。

队列和管道都是将数据存放在内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信的最佳选择。

我们应该尽量避免使用共享数据，尽可能的使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数据增多时，往往可以获得更好的可扩展性。

队列

队列彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。

创建队列的类（底层就是以管道和锁定的方式实现的）

Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递

参数介绍

maxsize是队列中允许最大项数，省略则无大小限制
但需要明确：
    1.队列内存放的是消息而非大数据
    2.队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

主要方法接收

q.put()方法用于插入数据到队列中
q.get()方法可以从队列读取并且删除一个元素

队列的使用

from multiprocessing import Process,Queue

# 存放数据
q = Queue(3)  # 在队列里不应该存放大的文件，因为占用的是内存
q.put(‘123‘)
q.put([1,2,3,])
q.put({‘name‘:‘xiaoyafei‘})

print(q.full())  # 可以查询是否已经满了

# 取数据
print(q.get())
print(q.get())
print(q.get())
print(‘Queue是否为空:‘,q.empty())  # 因为此时队列已经空了，所以接下来取数据就会堵塞
print(q.get())

# 运行结果为
True
123
[1, 2, 3]
{‘name‘: ‘xiaoyafei‘}
Queue是否为空: True  # 程序会在一直等待中
......

在这段代码中，我们可以发现：由Queue实例化出来的对象就是一个容器，而在我们存数据的时候就把数据丢到这个容器中，如果想要取数据的话就要到这个数据里面直接拿。

生产者消费者模型

为什么要使用生产者消费者模型

生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者生产的很快，而消费者处理的速度却很慢，那么生产者就必须等待消费者处理完后才能继续生产数据。同样的道理，如果消费者的处理速度大于生产者，那么消费者就需要等待生产者。为了解决这个问题于是引入了生产者和消费者模型。

什么是生产者和消费者模型

生产者消费者模式是通过一个容器来解决生产者和消费者的耦合问题的。生产者和消费者之间不直接通信，而通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列中去取，阻塞队列就相当于是一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。

关系图：https://www.processon.com/view/5b2cfb83e4b01a7cb45bafaf

生产者和消费者模型实现

import time
from multiprocessing import Process,Queue

def producer(q):  # 生产者
    for i in range(3):  # 3个人生产包子
        res = ‘包子%s‘%i
        time.sleep(1)
        print(‘生产者生产了%s‘%res)

        q.put(res)  # 生产完把包子丢到消息队列里面去

def consumer(q):  # 消费者
    while True:
        res = q.get()  # 从消息队列中取数据赋值给res
        time.sleep(2)
        print(‘消费者吃了%s‘%res)

if __name__ == ‘__main__‘:
    q = Queue()  # 如果不写大小，那么默认是无限制的

    # 生产者们
    p1 = Process(target=producer,args=(q,))
    p2 = Process(target=producer,args=(q,))
    p3 = Process(target=producer,args=(q,))

    # 消费者们
    c1 = Process(target=consumer,args=(q,))
    c2 = Process(target=consumer,args=(q,))

    p1.start()
    p2.start()
    p3.start()

    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()

    print(‘主进程‘)

讲解一下这段代码：

1.生产者来生产包子，消费者来吃包子。

2.生产者每1秒生产3个包子，消费者每2秒吃一个包子，备注为：每个生产者/每个消费者

3.创建消息队列Queue，无限制大小

4.p1/p2/p3.join()方法保证了子进程先运行完之后再运行主进程

5.生产者把包子丢到消息队列里面之后就不用管了

接下来，这段程序的运行结果为：

# 运行结果为
生产者生产了包子0
生产者生产了包子0
生产者生产了包子0
生产者生产了包子1
生产者生产了包子1
生产者生产了包子1
生产者生产了包子2
消费者吃了包子0
生产者生产了包子2
消费者吃了包子0
生产者生产了包子2
主进程
消费者吃了包子0
消费者吃了包子1
消费者吃了包子1
消费者吃了包子1
消费者吃了包子2
消费者吃了包子2
消费者吃了包子2
.......

# 但是程序在运行之后，会一直处于阻塞状态，因为消费者还在不停的取数据，但是生产者已经把数据生产完了

如何去解决上诉的问题呢？能不能思考，如果消费者去取数据的时候取了一个None，那么就停止：

import time
from multiprocessing import Process,Queue

def producer(q):  # 生产者
    for i in range(3):  # 3个人生产包子
        res = ‘包子%s‘%i
        time.sleep(1)
        print(‘生产者生产了%s‘%res)

        q.put(res)  # 生产完把包子丢到消息队列里面去

def consumer(q):  # 消费者
    while True:
        res = q.get()  # 从消息队列中取数据赋值给res
        if res == None:break  # 如果取的数据是空，那么就结束
        time.sleep(2)
        print(‘消费者吃了%s‘%res)

if __name__ == ‘__main__‘:
    q = Queue()  # 如果不写大小，那么默认是无限制的

    # 生产者们
    p1 = Process(target=producer,args=(q,))
    p2 = Process(target=producer,args=(q,))
    p3 = Process(target=producer,args=(q,))

    # 消费者们
    c1 = Process(target=consumer,args=(q,))
    c2 = Process(target=consumer,args=(q,))

    p1.start()
    p2.start()
    p3.start()

    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    q.put(None)  # 因为有两个消费者，那么就需要再往消息队列里面传递两个None
    q.put(None)
    print(‘主进程‘)

其实我们的思路无非就是发送结束信号而已，有另外一种队列提供了这种机制

JoinableQueue([maxsize])

这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享和条件变量实现的。

参数介绍

maxsize是队列中允许最大项数，省略则代表无大小限制

方法介绍

JoinableQueue的实例p除了与Queue对象相同的方法之外，还有：
1.q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发异常
2.q.join()：生产者使用此方法发出信号，直到队列中所有的项目都被处理。阻塞将持续到队列中的每个项目均调用q.task_done()方法为止

基于JoinableQueue实现生产者和消费者模型

import time
from multiprocessing import Process,JoinableQueue

def producer(q):  # 生产者
    for i in range(3):  # 3个人生产包子
        res = ‘包子%s‘%i
        time.sleep(2)
        print(‘生产者生产了%s‘%res)

        q.put(res)  # 生产完把包子丢到消息队列里面去
    q.join()  # 等待消息队列的数据都被取完

def consumer(q):  # 消费者
    while True:
        res = q.get()  # 从消息队列中取数据赋值给res
        time.sleep(1)
        print(‘消费者吃了%s‘%res)
        q.task_done()  # 消费者给生产者发送结束信号，但是还是在做q.get()


if __name__ == ‘__main__‘:
    q = JoinableQueue()  # 如果不写大小，那么默认是无限制的

    # 生产者们
    p1 = Process(target=producer,args=(q,))
    p2 = Process(target=producer,args=(q,))
    p3 = Process(target=producer,args=(q,))

    # 消费者们
    c1 = Process(target=consumer,args=(q,))
    c2 = Process(target=consumer,args=(q,))
    c1.daemon = True
    c2.daemon = True


    p1.start()
    p2.start()
    p3.start()

    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    print(‘主进程‘)