互斥锁，IPC队列

Posted 2020-10-04

进程同步（锁）

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,
part1：共享同一打印终端，发现会有多行内容打印到一行的现象（多个进程共享并抢占同一个打印终端，乱了）

#多进程共享一个打印终端(用python2测试看两个进程同时往一个终端打印,出现打印到一行的错误)
from multiprocessing import Process
import time
class Logger(Process):
    def __init__(self):
        super(Logger,self).__init__()
    def run(self):
        print(self.name)


for i in range(1000000):
    l=Logger()
    l.start()

View Code

part2：共享同一个文件，有的同学会想到，既然可以用文件共享数据，那么进程间通信用文件作为数据传输介质就可以了啊，可以，但是有问题：1.效率 2.需要自己加锁处理

#多进程共享一套文件系统
from multiprocessing import Process
import time,random

def work(f,msg):
    f.write(msg)
    f.flush()


f=open(‘a.txt‘,‘w‘) #在windows上无法把f当做参数传入，可以传入一个文件名，然后在work内用a+的方式打开文件，进行写入测试
for i in range(5):
    p=Process(target=work,args=(f,str(i)))
    p.start()

View Code

需知：加锁的目的是为了保证多个进程修改同一块数据时，同一时间只能有一个修改，即串行的修改，没错，速度是慢了，牺牲了速度而保证了数据安全。

进程之间数据隔离，但是共享一套文件系统，因而可以通过文件来实现进程直接的通信，但问题是必须自己加锁处理

所以，就让我们帮文件当做数据库，模拟抢票（Lock互斥锁）

from multiprocessing import Process,Lock
import time,random,os
import json

def tar(mul):
    dic1 = json.load(open(‘db.txt‘,))
    print("还剩%s张票"%dic1[‘count‘],os.getpid())
    if dic1[‘count‘]>0:
        time.sleep(random.randint(1,10))
        mul.acquire()
        dic = json.load(open(‘db.txt‘, ))
        if dic[‘count‘] > 0:
            dic[‘count‘] -= 1
            json.dump(dic,open(‘db.txt‘,‘w‘))
            print(‘购票成功‘,os.getpid())
        mul.release()

if __name__==‘__main__‘:
    mul= Lock()
    for i in range(20):
        p = Process(target=tar,args=(mul,))
        p.start()

View Code

 进程间通信（IPC）方式一：队列（推荐使用）

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

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

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

    参数介绍：

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

　 方法介绍：

    主要方法：

1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。
如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。
如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
3  
4 q.get_nowait():同q.get(False)
5 q.put_nowait():同q.put(False)
6 
7 q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。
8 q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。
9 q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样
了解：

1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞

2 q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。
关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。
3 q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

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

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

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

基于队列实现生产者消费者模型

 

创建队列的另外一个类：

    JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

    参数介绍：

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

　 方法介绍：

    JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

    q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

    q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

import os,time,random
from multiprocessing import  Process,JoinableQueue

def producer_dup(q):
    for i in range(10):
        time.sleep(2)
        print(os.getpid(),‘制造了包子\033[45m%s\033[0m‘%i)
        q.put(‘包子%s‘%i)
    q.join()
def producer_gu(q):
    for i in range(3):
        time.sleep(2)
        print(os.getpid(),‘制造了骨头\033[45m%s\033[0m‘%i)
        q.put(‘骨头%s‘%i)
    q.join()
def producer_sh(q):
    for i in range(3):
        time.sleep(2)
        print(os.getpid(),‘制造了泔水\033[45m%s\033[0m‘%i)
        q.put(‘泔水%s‘%i)
    q.join()
def constumer(q):
    while 1:
        time.sleep(random.randint(1,3))
        ret = q.get()
        print(os.getpid(),‘吃了\033[44m%s\033[0m‘%ret)
        q.task_done()


if __name__ == ‘__main__‘:
    q = JoinableQueue()
    p1 = Process(target=producer_dup,args=(q,))
    p2 = Process(target=producer_gu,args=(q,))
    p3 = Process(target=producer_sh,args=(q,))
    p4 = Process(target=constumer,args=(q,))
    p5 = Process(target=constumer,args=(q,))
    p4.daemon=True
    p5.daemon=True
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p5.start()
    p1.join()
    p2.join()
    p3.join()
    print(‘end‘)