首页 > > 网络编程 > 其它 >

进程同步控制

2018-12-02 06:16:25来源：博客园阅读 ()

锁——multiprocessing.Lock

什么是锁？

当多个进程使用同一份数据资源的时候，会引发数据安全或顺序混乱问题。这个时候我们希望进程可以一个一个的去获取和修改数据，将几个并发的进程编程串行，这样就可以保证数据的安全。我们可以引用Lock模块来帮我们来实现将异步执行的程序在加锁的代码段中同步去执行。

mport 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()

多进程抢占输出资源

# 由并发变成了串行,牺牲了运行效率,但避免了竞争
import os
import time
import random
from multiprocessing import Process,Lock

def work(lock,n):
    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(3):
        p=Process(target=work,args=(lock,i))
        p.start()

使用锁维护执行顺序

使用锁来维护执行顺序

上面这种情况虽然使用加锁的形式来实现了顺序的执行，但是程序由并发编程串行了。这样虽然降低了效率，却保证了数据的安全。

模拟抢票，来看看数据安全的重要性。

#文件db的内容为：{"count":1}
#注意一定要用双引号，不然json无法识别
#并发运行，效率高，但竞争写同一文件，数据写入错乱
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db'))
    time.sleep(0.1) #模拟读数据的网络延迟
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模拟写数据的网络延迟
        json.dump(dic,open('db','w'))
        print('\033[43m购票成功\033[0m')

def task():
    search()
    get()

if __name__ == '__main__':
    for i in range(100): #模拟并发100个客户端抢票
        p=Process(target=task)
        p.start()

多进程同时抢购余票

#文件db的内容为：{"count":5}
#注意一定要用双引号，不然json无法识别
#并发运行，效率高，但竞争写同一文件，数据写入错乱
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db'))
    time.sleep(random.random()) #模拟读数据的网络延迟
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(random.random()) #模拟写数据的网络延迟
        json.dump(dic,open('db','w'))
        print('\033[32m购票成功\033[0m')
    else:
        print('\033[31m购票失败\033[0m')

def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()

if __name__ == '__main__':
    lock = Lock()
    for i in range(100): #模拟并发100个客户端抢票
        p=Process(target=task,args=(lock,))
        p.start()

使用锁来保证数据安全

总结：

加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改。牺牲效率却保证了数据安全。

虽然可以用文件共享数据实现进程间通信：

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

因此我们最好寻找一种解决方案能够兼顾：

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

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

队列和管道都是将数据存放在内存中

队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来。

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

信号量——multiprocessing.Semaphore(了解)

信号量概念介绍:

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

代码示例:

from multiprocessing import Process,Semaphore
import time,random

def go_ktv(sem,user):
    sem.acquire()
    print('%s 占到一间ktv小屋' %user)
    time.sleep(random.randint(0,3)) #模拟每个人在ktv中待的时间不同
    sem.release()

if __name__ == '__main__':
    sem=Semaphore(4)
    p_l=[]
    for i in range(13):
        p=Process(target=go_ktv,args=(sem,'user%s' %i,))
        p.start()
        p_l.append(p)

    for i in p_l:
        i.join()
    print('============》')

例子

信号量代码示例

事件——multiprocessing.Event(了解)

概念及方法介绍:

python进程的事件主要提供了四个方法:

set(), wait(), clear(), is_set()

事件处理的机制: 全局定义了一个"Flag", 如果"Flag"值为False, 那么当程序执行event.wait()方法时就会阻塞, 如果"Flag"值为Ture, 那么event.wait()方法时便不再阻塞.

事件的初始"Flag"状态为False

clear: 将"Flag"设置为False

set: 将"Flag"设置为True

is_set: 查看"Flag"状态

from multiprocessing import Process, Event
import time, random


def car(e, n):
    while True:
        if not e.is_set():  # 进程刚开启，is_set()的值是Flase，模拟信号灯为红色
            print('\033[31m红灯亮\033[0m，car%s等着' % n)
            e.wait()    # 阻塞，等待is_set()的值变成True，模拟信号灯为绿色
            print('\033[32m车%s 看见绿灯亮了\033[0m' % n)
            time.sleep(random.randint(3, 6))
            if not e.is_set():   #如果is_set()的值是Flase，也就是红灯，仍然回到while语句开始
                continue
            print('车开远了,car', n)
            break


def police_car(e, n):
    while True:
        if not e.is_set():# 进程刚开启，is_set()的值是Flase，模拟信号灯为红色
            print('\033[31m红灯亮\033[0m，car%s等着' % n)
            e.wait(0.1) # 阻塞，等待设置等待时间，等待0.1s之后没有等到绿灯就闯红灯走了
            if not e.is_set():
                print('\033[33m红灯,警车先走\033[0m，car %s' % n)
            else:
                print('\033[33;46m绿灯，警车走\033[0m，car %s' % n)
        break



def traffic_lights(e, inverval):
    while True:
        time.sleep(inverval)
        if e.is_set():
            print('######', e.is_set())
            e.clear()  # ---->将is_set()的值设置为False
        else:
            e.set()    # ---->将is_set()的值设置为True
            print('***********',e.is_set())


if __name__ == '__main__':
    e = Event()
    for i in range(10):
        p=Process(target=car,args=(e,i,))  # 创建是个进程控制10辆车
        p.start()

    for i in range(5):
        p = Process(target=police_car, args=(e, i,))  # 创建5个进程控制5辆警车
        p.start()
    t = Process(target=traffic_lights, args=(e, 10))  # 创建一个进程控制红绿灯
    t.start()

    print('============》')

红绿灯实例