python-互斥锁和进程同步

作者：马育民 • • 阅读：10412

概述

通过 python-多进程修改公共变量引发的血案 可见，多进程修改公共变量时，会引发bug，为了解决bug，要用到进程同步

在多进程修改公共变量时，如下图：

  1. 先让进程0读变量num的值
    注意：具体是哪个进程不一定，由操作系统调度，图中由 进程1 执行
  2. 进程0计算后将值写入内存
  3. 然后进程1读变量num的值
  4. 进程1计算后将值写入内存

让进程之间有顺序的执行，互相合作，互相等待

临界资源

某些公共资源只能一个进程访问操作，这样的资源称为临界资源（互斥资源，共享变量）

如：上面的商品数量、票号，或者打印机

临界区

访问临界资源的那段代码称为临界区

如：上面访问商品数量、票号的代码

进程互斥

由于多进程要共享资源，而有些资源需要互斥使用，因此各进程间竞争使用这些资源

如：多进程修改公共变量，多进程访问打印机等

进程同步

多进程访问操作公共资源时，需要多个进程有顺序的执行，互相合作、互相等待，共同完成一项任务。

一般来说，同步往往包含互斥

使用互斥锁

为了实现多进程同步和互斥，提出 互斥锁，
当多个进程执行到 申请锁 代码时，会竞争，只有申请到锁的进程才会继续向下运行，其他进程都阻塞（暂停），直到该进程 释放锁 时，其他进程才会继续执行

注意：
申请锁 和 释放锁 之间的代码，只有一个进程在运行，所以互斥锁不要滥用，会导致多进程效率低下

申请锁和释放锁的位置

在临界区前加 申请锁，在临界区之后加 互斥锁

Lock类的使用说明

multiprocessing 模块中定义了Lock类，实现互斥锁，让进程同步

注意：
当多个进程 修改 公共变量时，才需要互斥锁

官方文档链接：
https://docs.python.org/3/library/multiprocessing.html?highlight=multiprocessing#multiprocessing.Lock

导入模块

from multiprocessing import Lock

创建锁

lock=multiprocessing.Lock()

申请锁

表示之后的代码全部被锁定，只能一个进程执行

lock.acquire(block=True, timeout=None)

参数：

  • block：

    • 为True（默认值），最先执行此方法的进程 获得“锁”，返回True， 然后只有该进程继续向下执行，其它进程执行此方法时会 阻塞，直到该进程 释放锁 为止。返回True。
    • 为False，最先执行此方法的进程 获得“锁”，返回True，该进程继续向下执行。虽然有进程处于锁定状态，但其它进程执行该方法时 不会阻塞，返回False，也会继续向下执行

  • timeout：阻塞超时时间。当 block为True时，设置该参数 有意义

    • 正浮点数：如果有进程 “获得锁”，其他进程执行该方法时，最多阻塞指定的秒数，返回False，并继续向下执行
    • 负值和零：如果有进程 “获得锁”，其他进程执行该方法时，最多阻塞0秒，返回False，并继续向下执行
    • None（默认值）：如果有进程 “获得锁”，其他进程执行该方法时，一直阻塞

释放锁

释放锁。获得锁的进程，才能执行该方法，其它处于lock.acquire()方法阻塞的进程，可以继续执行

lock.release()

案例1-加锁

#coding=utf-8
from multiprocessing import Process,Value,Lock
import time,os

#双十一秒杀

def buy(n,index,lock):
    lock.acquire()
    print('%d号买家-开始购买,pid=%d,num=%d'%(index,os.getpid(),n.value))
    #修改变量
    if n.value>0:
        # time.sleep(10)
        n.value-=1
    else:
        print('售罄')
    print('%d号买家-结束购买,pid=%d,num=%d'%(index,os.getpid(),n.value))
    lock.release()

if __name__ == "__main__":
    stime=time.time()
    #创建锁
    lock=Lock()
    #秒杀商品数量
    num=Value('i',100)
    ps=[]
    for index in range(101):
        p=Process(target=buy,args=(num,index,lock))
        ps.append(p)

    for index in range(101):
        ps[index].start()

    for index in range(101):
        ps[index].join()

    #主进程打印值
    print('主进程，num=%s'%num.value)
    etime=time.time()
    print(etime-stime)

加了互斥锁后，执行顺序如下图：

进程访问公共变量前申请锁，此时其他进程运行到此代码时会阻塞，不再执行，该进程修改公共变量后释放锁，其他进程继续执行

修改锁的位置

将上例中buy()函数的申请锁、释放锁的位置做调整，如下代码：

def buy(n,index,lock):
    print('%d号买家-开始购买,pid=%d,num=%d'%(index,os.getpid(),n.value))
    lock.acquire()
    #修改变量
    if n.value>0:
        # time.sleep(10)
        n.value-=1
    else:
        print('售罄')
    lock.release()
    print('%d号买家-结束购买,pid=%d,num=%d'%(index,os.getpid(),n.value))

分析打印的结果，然后执行观察结果

lock支持with语句

#coding=utf-8
from multiprocessing import Process,Value,Lock
import time
import os

#双十一秒杀

def buy(n,lock,index):
    #进程锁with语句
    with lock:
        print('%d号乘客-开始买票,pid=%d,num=%d'%(index,os.getpid(),n.value))
        #修改变量
        if n.value>=1:
            #time.sleep(10)
            n.value-=1
        else:
            print('没有票了')
        print('%d号乘客-结束买票,pid=%d,num=%d'%(index,os.getpid(),n.value))


if __name__ == "__main__":
    stime=time.time()
    #秒杀商品数量
    num=Value('i',100)
    lock = Lock()
    ps=[]
    for index in range(101):
        p=Process(target=buy,args=(num,lock,index))
        ps.append(p)

    for index in range(101):
        ps[index].start()

    for index in range(101):
        ps[index].join()


    #主进程打印值
    print('主进程，num=%s'%num.value)
    etime=time.time()
    print(etime-stime)

案例2-修改加锁（错误）

#coding=utf-8
from multiprocessing import Process,Value,Lock
import time
import os

#票总数
count=500
#窗口数量
windownum=3
def sell(n,index,lock):
    lock.acquire()
    print('%d号售票口-开始售票,票号：%d,pid=%d'%(index,n.value,os.getpid()))
    #修改变量
    while n.value<=count:
        print('%d号售票口-售出票号：%d,pid=%d'%(index,n.value,os.getpid()))
        i=n.value+1
        # time.sleep(1)
        n.value=i
        print('%d号售票口-下一票号：%d,pid=%d'%(index,n.value,os.getpid()))
    else:
        print('%d号售票口-售罄'%(index))
    lock.release()
if __name__ == "__main__":
    stime=time.time()
    #初始化票号码
    num=Value('i',1)
    ps=[]
    lock=Lock()
    for index in range(windownum):
        p=Process(target=sell,args=(num,index,lock))
        ps.append(p)

    for index in range(windownum):
        ps[index].start()

    for index in range(windownum):
        ps[index].join()

    #主进程打印值
    print('主进程，num=%s'%num.value)
    etime=time.time()
    print(etime-stime)

执行结果如下图：

一直都是0号售票口在执行，原因是加锁的位置错误，在循环之前加锁，在循环之后释放锁，所以只有一个进程在运行

案例2-修改加锁（正确）

#coding=utf-8
from multiprocessing import Process,Value,Lock
import time
import os

#票总数
count=500
#窗口数量
windownum=3
def sell(n,index,lock):
    print('%d号售票口-开始售票,票号：%d,pid=%d'%(index,n.value,os.getpid()))
    #修改变量
    while True:
        lock.acquire()
        if n.value<=count:
            print('%d号售票口-售出票号：%d,pid=%d'%(index,n.value,os.getpid()))
            i=n.value+1
            # time.sleep(0.1)
            n.value=i
            print('%d号售票口-下一票号：%d,pid=%d'%(index,n.value,os.getpid()))
            lock.release()
        else:
            print('%d号售票口-售罄'%(index))
            lock.release()
            break

if __name__ == "__main__":
    stime=time.time()
    #初始化票号码
    num=Value('i',1)
    ps=[]
    lock=Lock()
    for index in range(windownum):
        p=Process(target=sell,args=(num,index,lock))
        ps.append(p)

    for index in range(windownum):
        ps[index].start()

    for index in range(windownum):
        ps[index].join()

    #主进程打印值
    print('主进程，num=%s'%num.value)
    etime=time.time()
    print(etime-stime)

总结：

多进程 并发修改 同一个公共变量（或文件）时，一般需要多进程 同步 运行，需要加 互斥锁，加锁的位置要注意，一般在 临界区 前后 加锁 （也要结合具体情况，具体分析），加锁的区域内就是 单进程 在运行，效率会降低

如果多进程 没有 并发修改同一个公共变量（或文件）时，一般 不需要加锁

原文出处：/show_1EF2ozhHoCsp.html

吉ICP备15003043号