python全栈学习--day39(multiprocess模块)

Posted 2020-11-04 鎵譹_John

 

进程同步(multiprocess.Lock、multiprocess.Semaphore、multiprocess.Event)

锁 —— multiprocess.Lock

      通过刚刚的学习，我们千方百计实现了程序的异步，让多个任务可以同时在几个进程中并发处理，他们之间的运行没有顺序，一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用IO资源，但是也给我们带来了新的问题。

　　当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题。

实例一：

import os
import time
import random
from multiprocessing import Lock     # 锁
from multiprocessing import Process

def work(n,lock):
    lock.acquire()
    print(\'%s: %s is running\' %(n,os.getpid()))
    time.sleep(random.random())
    print(\'%s:%s is done\' %(n,os.getpid()))
    lock.release()

if __name__ == \'__main__\':
    lock = Lock()
    for i in range(10):
        p=Process(target=work,args=(i,lock))
        p.start()

　执行输出：

说明：上面当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题。

而这个时候就用到了锁的机制

 

import os
import time
import random
from multiprocessing import Process

def work(n):
    print(\'%s: %s is running\' %(n,os.getpid()))
    time.sleep(random.random())
    print(\'%s:%s is done\' %(n,os.getpid()))

if __name__ == \'__main__\':
    for i in range(3):
        p=Process(target=work,args=(i,))
        p.start()

　　执行输出：

锁的理论

acquire 都想要钥匙开锁

最先拿到钥匙的，它会做几件事情：

1.用钥匙开门

2.把钥匙带进门里

3.把门反锁

 

执行了acpuire，表示想要钥匙

其他9个人想要钥匙，但是进不去呀

必须等到那个人出来

release 表示开门，从门里面出来

把钥匙挂在门上

示例二：

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

def work(lock,n):
    lock.acquire()
    print(\'%s: %s is running\'%(n,os.getppid()))
    time.sleep(random.random())
    print(\'%s:%s is done\'%(n,os.getppid()))
    lock.release()

if __name__==\'__main__\':
    lock = Lock()
    for i in range(3):
        p = Process(target=work,args=(lock,i))
        p.start()

上面这种情况虽然使用加锁的形式实现了顺序的执行，但是程序又重新变成串行了，这样确实会浪费了时间，却保证了数据的安全。

　　接下来，我们以模拟抢票为例，来看看数据安全的重要性。 

　　

模拟数据库，创建一个文件ticket，内容如下：

1	{"count":1}

注意一定要用双引号，不然json无法识别

程序1

import json
from multiprocessing import Process

def check_ticket(i):
    with open(\'ticket\')as f:
        ticket_count = json.load(f)
    print(\'person%s\'%i,ticket_count[\'count\'])

if __name__ == \'__main__\':
    for i in range(5):
        Process(target=check_ticket,args=(i,)).start()

　执行输出：

开始买票，买票的时候，每个人的网络环境不同(带宽速率，地域....)会有延迟，这样服务器会把最先收到的网络请求后处理好的数据给予最先发送的client。

服务器请求原理：

收到请求之后，从数据库中读取数据
当你发票还有余票时，把票减少这件事情记录下来
中间会经历网络延时

 

　

import json
import time
import random
from multiprocessing import Process
 
def buy_ticket(i):  # 购票
    with open(\'ticket\') as f:  # 读取文件
        tick_count = json.load(f)  # 反序列化
        time.sleep(random.random())  # 读取延时
    if tick_count[\'count\'] > 0:  # 当余票小于0时
        print(\'person%s购票成功\'%i)
        tick_count[\'count\'] -= 1  # 票数减1
    else:
        print(\'余票不足,person%s购票失败\'%i)
    time.sleep(random.random())  # 写入延时
    with open(\'ticket\',\'w\') as f:
        json.dump(tick_count,f)  # 写入文件
 
if __name__ == \'__main__\':
    for i in range(5):
        Process(target=buy_ticket,args=(i,)).start()

　

模拟这2次延时
是因为服务器和数据库不在一台机器上面，它们之间交互数据，必然有延时

上面的代码只模拟了读取延时和写入延时，没有模拟请求延时

执行时间：

从输出结果上来看，库存只有1张票，但是5个人都购票成功了。这显然是不合理的！

这就造成了数据不安全

怎么解决? 当然是加锁啦

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

def buy_ticket(i,lock): #购票
    lock.acquire()      #取得锁
    with open(\'ticket\')as f:    #读取文件
        tick_count = json.load(f)   #反序列化
        time.sleep(random.random()) #读取延时
    if tick_count[\'count\'] >0:      #当余票小于0时
        print(\'person%s购票成功\' % i)
        tick_count[\'count\'] -= 1  # 票数减1
    else:
        print(\'余票不足,person%s购票失败\' % i)
    time.sleep(random.random())  # 写入延时
    with open(\'ticket\', \'w\') as f:
        json.dump(tick_count, f)  # 写入文件
    lock.release()  # 释放锁


if __name__ == \'__main__\':
    lock = Lock()  # 创建锁
    for i in range(5):  # 模拟5个用户抢票
        Process(target=buy_ticket, args=(i, lock)).start()

　　

从结果上来看，是正确的。保证了数据的安全性。

查数据，不涉及数据安全，因为没有修改。

总结：

#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理

#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

　

信号量 —— multiprocess.Semaphore                

　

互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。
假设商场里有4个迷你唱吧，所以同时可以进去4个人，如果来了第五个人就要在外面等待，等到有人出来才能再进去玩。
实现：
信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。
信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念

　上面讲的锁是一个，现在锁变成一串了，

数量由你来控制

下面有一个小KTV，只能容纳4个人，第5个人，就没有钥匙了

先不用信号量

1
2
3
4
5
6
7
8
9
10
11
12
13
#KTV 4个人
import time
import random
from multiprocessing import Process,Semaphore
 
def ktv(i):
    print(\'person %s 进来唱歌了\'%i)
    time.sleep(random.randint(1,5))
    print(\'person %s 从ktv出去了\'%i)
 
if __name__ == \'__main__\':
    for i in range(6):
        Process(target=ktv,args=(i,)).start()
执行输出：

person 0 进来唱歌了
person 1 进来唱歌了
person 2 进来唱歌了
person 3 进来唱歌了
person 4 进来唱歌了
person 5 进来唱歌了
person 3 从ktv出去了
person 4 从ktv出去了
person 2 从ktv出去了
person 1 从ktv出去了
person 0 从ktv出去了
person 5 从ktv出去了

结果是有问题的，5个人，都可以进去

使用信号量来实现

　　

import time
import random
from multiprocessing import Process, Semaphore


def ktv(i,sem):
    sem.acquire()
    print(\'person %s 进来唱歌了\' % i)
    time.sleep(random.randint(1, 5))
    print(\'person %s 从ktv出去了\' % i)
    sem.release()

if __name__ == \'__main__\':
    sem = Semaphore(4)  #初始化信号量，数量为4
    for i in range(6):  #模拟5个人
        Process(target=ktv,args=(i,sem)).start()

　　

在同一时间，最多有4个人进去

acquire()是一个阻塞行为

 

信号量和锁有点类似
那么它们之间的区别在于：
信号量，相当于计数器

它是锁+计数器
调用acquire() 计数器-1
当计数器到 0 时，再调用 acquire() 就会阻塞，直到其他线程来调用release()
调用release() 计数器+1

事件 —— multiprocess.Event

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

    事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

clear：将“Flag”设置为False
set：将“Flag”设置为True

著名的计算机模型：红绿灯
车 比方是一个进程，wait() 等红灯
根据状态变化，wait遇到true信号，就非阻塞
遇到False，就阻塞
交通灯 也是有一个进程 红灯->False 绿灯->True

这里没有黄灯

事件有几个方法：

wait的方法 根据一个状态来决定自己是否要阻塞
状态相关的方法
　　set 将状态改为True
　　clear 将状态改为False
　　is_set 判断当前的状态是否为True

from multiprocessing import Event

e = Event()         #创建一个事件的对象
print(e.is_set())   #在事件的创世之初，状态为False

　执行输出：False

在看一个列子

1 2 3 4 5 6

from multiprocessing import Event   e = Event() #创建一个事件的对象 print(e.is_set())  # 在事件的创世之初,状态为False e.wait() print(\'1\')

执行输出： False，然后程序一致卡着

为啥呢？因为状态值为 False，那么当程序执行 event.wait 方法时就会阻塞