互斥锁、共享内存方式以及生产者消费者模型

女爷i 2021-12-16 11:53 213阅读 0赞

# 守护进程 #

1、守护进程的概念

进程指的是一个正在运行的程序，守护进程也是一个普通进程

意思就是一个进程可以守护另一个进程

import time
    from multiprocessing import Process
    
    def task():
        print("子进程   走起！！！")
        time.sleep(5)
        print("子进程  挂掉了")
    
    if __name__ == '__main__':
        print("主进程   走起！！！")
    
        p = Process(target=task)
        p.daemon = True  # 将子进程设置为守护进程
        p.start()
        time.sleep(3)
        print("主进程  挂掉了")
    # --1）如果设置为守护进程，那么主进程一旦结束，那么子进程立马也结束（代码终止执行）
    # --2）如果没有设置为守护进程，那么主进程结束执行完代码之后，子进程还会继续执行子进程里面的代码

2、结论

如果a 是 b 的守护进程，那么 b 就是被守护的进程， b要是（代码运行完毕）死  ，a也会跟着死

守护进程在主进程代码运行结束之后就死了

3、使用场景

父进程交给了子进程一个任务，任务还没有完成父进程就结束了，子进程就没有继续执行的意义了

例如qq接到一个视频文件，于是开启了一个子进程来下载，如果中途退出了qq，下载任务就没必要继续执行了

# 互斥锁（重点） #

### 为什么需要互斥锁 ###

当我们同时开启几个子进程的时候，而这几个子进程又要同时操作一个资源（文件或者控制台）,

将会导致数据的错乱问题（谁先抢到谁打印，数据混乱看不懂）

### 解决方案1： ###

加join    （就是每开一个子进程，就加一个join，限制后面的子进程不能开启）

弊端：

--1、把原本并发的任务变成了串行，避免了数据错乱问题，但是效率降低了，

而且是执行完第一个子进程才开启第二个子进程，都放在主程序也是同样效果，没必要开子进程

--2、原本多个进程之间公平竞争，join执行的顺序是定死的，不合理

### 解决法案2： ###

给公共资源加锁（互斥锁）

互斥锁，字面理解就是互相排斥的锁，在程序中指的是，如果这个资源已经被锁住了，其他进程就无法使用

需要强调的是：锁，并不是真正的把资源锁起来，只是在代码里加限制，限制你的代码不能执行

import time,random
    from multiprocessing import Process,Lock
    
    def task1(lock):
        lock.acquire()      # 上锁
        print('任务一  走起')
        time.sleep(random.randint(0,2))
        print('任务一  结束')
        lock.release()      # 解锁
    
    def task2(lock):
        lock.acquire()
        print('任务二  走起')
        time.sleep(random.randint(0,2))
        print('任务二  结束')
        lock.release()
    
    def task3(lock):
        lock.acquire()
        print('任务三  走起')
        time.sleep(random.randint(0,2))
        print('任务三  结束')
        lock.release()
    
    if __name__ == '__main__':
        lock = Lock()
        p1 = Process(target=task1,args=(lock,))
        p2 = Process(target=task2,args=(lock,))
        p3 = Process(target=task3,args=(lock,))
        p1.start()
        p2.start()
        p3.start()

注意：

--1、不要对同一代码执行多次acquire，会锁死程序无法执行，一次acquire必须对应一次release

（一次加锁相当于在代码做一次判断，判断成功改状态，再加锁又相当于做一次判断，判断就会失败，锁死）

--2、想要保证数据安全，必须保证所有进程使用同一把锁

（如果每个进程拿到的锁都不一样，那么判断的条件就不会同步）

### 锁和join的区别 ###

1、join是固定了执行顺序，会造成父进程等待子进程

锁依然是公平竞争，谁先抢到谁先执行，父进程也可以执行其他任务

2、最主要的区别：

join是把进程的所有任务全部串行

锁可以所以任意代码，一行也可以，可以自己调整粒度  粒度(指的是被锁住代码的大小，越小效率越快)

# IPC #

IPC就是进程间通讯，通讯指的是交换数据

进程之间的内存是相互隔离的，当一个进程想要把数据给另外一个进程，就需要考虑到IPC

IPC方式：

管道：只能单向通讯(一边读，一边写)，数据都是二进制

文件：在硬盘上创建共享文件

优点：硬盘的空间大，所以共享的数据量几乎没有限制

缺点：硬盘读取速度慢

socket：编程复杂度高

共享内存： 必须由操作系统来分配     （必须掌握）

优点：速度快

缺点：数据量不能太大

# 共享内存的方式 #

### 1、Manager类 ###

Manager提供了很多数据结构，如list，dict等等（可以在源码中看到）

Manager所创建出来的数据结构，具备进程间共享的特点

from multiprocessing import Manager,Process,Lock
    import time
    
    def task(dic,lock):
        lock.acquire()  # 这里如果不加锁的话，可能出现多个进程刚取出来的值都是100，然后修改完都是99
        nums = dic["nums"]
        # time.sleep(2)  # 所有进程拿到100睡2秒，最后修改的数据就是99
        dic["nums"] = nums - 1
        lock.release()
    if __name__ == '__main__':
        lock = Lock()
        m = Manager()    # 创建一个manager对象
        dic = m.dict({
        "nums":100})    # 括号里需要的是一个映射关系，帮我们生成一个进程间共享的字典
        for i in range(5):
            p = Process(target=task,args=(dic,lock))
            p.start()
        time.sleep(10)
        print(dic["nums"])

需要注意的是：manager创建的一些数据结构是不带锁的，可能会出现同时修改一个数据的情况，会出问题，不推荐使用、

### 2、Queue队列 帮我们处理了锁的问题 （重点） ###

队列是一种特殊的数据结构，先存储的先取出，就像排队，先进的先出

相反的就是堆栈，先存取的后取出，就像把衣服叠进箱子，取的时候先取后叠进去的

函数嵌套调用时，执行的顺序也是先进后出，也称之为函数栈

from multiprocessing import Queue
    
    q = Queue(maxsize=3)  # 创建一个队列，里面的参数表示的是队列的容量，如果不写默认无限大
    
    # 存数据 q.put() 放进去
    q.put("a")
    q.put("b")
    q.put("c")
    # q.put("d")
    # 如果队列里面已经满了，你还继续往里放，他就会进入阻塞状态，一直等到队列有空位置把数据放进去
    
    # 取数据 q.get() 拿出来
    print(q.get())
    print(q.get())
    print(q.get())
    # print(q.get())
    # 同样的，如果队列里面已经空了，你还继续取，它也会进入阻塞状态，一直等到队列里有数据然后取出来
    
    q.get(block=True,timeout=2)
    # 里面block参数表示的是是否阻塞，默认True（阻塞）,当设置为False时，并且队列为空时会立马抛出异常
    q.put("aaa",block=True,timeout=2)
    # 里面block参数表示的是是否阻塞，默认True（阻塞）,当设置为False时，并且队列为满时会立马抛出异常
    # timeout 表示的时阻塞的超时时间，超过时间还是没有位置或着空的话就会抛出异常

# 生产者消费者模型 （重点） #

### 1、概念 ###

模型  就是解决某个问题的套路

产生数据的一方称之为生产者

处理数据的一方称之为消费者

例如：饭店厨师就是生产者，吃饭的人就是消费者

### 2、生产者和消费者产生的问题 ###

生产者和消费者，处理速度不平衡，一方快一方慢，导致一方需要等待另一方再能接着往下执行

### 3、生产者消费者模型解决这个问题的思路 ###

原本，双方是耦合在一起，消费者必须等待生产着生成完毕再开始处理，反过来，

如果消费者消费速度太慢，生产着必须等待其处理完毕才能开始生成下一个数据

### 4、解决的方案 ###

将双方分开来，一方专门，负责生产，一方专门负责处理

一样一来，双方的数据就不能直接交互了，双方需要共同的容器

这样就解决了双方能力不平衡的问题，做的快的一方可以继续做，不需要等待另一方，提高了整体的运行效率

from multiprocessing import Process,Queue
    import time,random
    def eater(q):
        for i in range(5):
            time.sleep(random.randint(0, 2))
            food = q.get()
            print("%s吃完了"%food)
    
    def cooker(q):
        for i in ["包子","骨头","土","面","刀子"]:
            time.sleep(random.randint(0, 2))
            q.put(i)
            print("%s做好了"%i)
    
    if __name__ == '__main__':
        q = Queue()
        cp = Process(target=cooker,args=(q,))
        ep = Process(target=eater,args=(q,))
        cp.start()
        ep.start()

转载于:https://www.cnblogs.com/hesujian/p/10968848.html