Python的进程锁，进程队列-阿里云开发者社区

锁 lock 互斥锁

什么是Python进程锁？Python进程锁如何创建和关闭

进程是系统进行资源分配和调度的基本单位，当一个python程序在运行时就会给它分配单个或者是多个进程来利用资源。那想在python中将某一个任务进程锁住不让它被其他对象访问的话就要用到进程锁了，下面来给大家介绍python进程锁的含义和使用方法。

from multiprocessing import Process,Lock

上锁和解锁是一对, 连续上锁不解锁是死锁 ,只有在解锁的状态下,其他进程才有机会上锁

锁的方法：原码

lock同一时间只能有一个进程上锁，保证进程同步

#创建一把锁，进程锁

lock = Lock()

#上锁

lock.acquire()

#lock.acquire() # 连续上锁,造成了死锁现象;

print("我在袅袅炊烟 … 你在焦急等待 … 厕所进行时 … ")

#解锁

lock.release()

上锁和解锁必须成对存在，如果上两把锁，下面的代码会不执行，一直等待解锁。代码阻塞在232行。这种现象叫做死锁

上锁，解锁，上锁，解锁，不影响程序运行

模拟12306 抢票软件

import json,time,random

# 1.读写数据库当中的票数
def wr_info(sign , dic=None):
    if sign == "r":
        with open("ticket",mode="r",encoding="utf-8") as fp:
            dic = json.load(fp)
        return dic
        
    elif sign == "w":
        with open("ticket",mode="w",encoding="utf-8") as fp:
            json.dump(dic,fp)

# dic = wr_info("w",dic={"count":0})
# print(dic , type(dic) )

# 2.执行抢票的方法
def get_ticket(person):
    # 先获取数据库中实际票数
    dic = wr_info("r")
    
    # 模拟一下网络延迟
    time.sleep(random.uniform(0.1,0.7))
    
    # 判断票数
    if dic["count"] > 0:
        print("{}抢到票了".format(person))
        # 抢到票后,让当前票数减1
        dic["count"] -= 1
        # 更新数据库中的票数
        wr_info("w",dic)
    else:
        print("{}没有抢到票哦".format(person))
        
# 3.对抢票和读写票数做一个统一的调用
def main(person,lock):
    
    # 查看剩余票数
    dic = wr_info("r")
    print("{}查看票数剩余: {}".format(person,dic["count"]))
    
    # 上锁
    lock.acquire()
    # 开始抢票
    get_ticket(person)
    # 解锁 
    lock.release()
    
if __name__ == "__main__":
    lock = Lock()
    lst = ["梁新宇","康裕康","张保张","于朝志","薛宇健","韩瑞瑞","假摔先","刘子涛","黎明辉","赵凤勇"]
    for i in lst:
        p = Process(    target=main,args=(  i  , lock  )   )
        p.start()

创建进程,开始抢票是异步并发程序

直到开始抢票的时候,变成同步程序,

先抢到锁资源的先执行,后抢到锁资源的后执行;

按照顺序依次执行;是同步程序;

“”"

CPU异步，延迟不一样，可能导致抢到的顺序不一样

涉及到改数据时，就要上锁，保证进程同步，确保数据安全。不能用异步

既然进程之间不会共享资源，那么进程之间怎么可以共享锁呢？

进程之间锁实际上也是隔离的，lock的底层通过socket给各进程发消息，通过收发消息，让各进程得到锁的状态，变相实现锁在各个进程间共享

信号量 Semaphore

本质上就是锁,只不过是多个进程上多把锁,可以控制上锁的数量

Semaphore = lock + 数量

from multiprocessing import Semaphore , Process
import time , random

    # 同一时间允许多个进程上5把锁
    sem = Semaphore(5)
    #上锁
    sem.acquire()
    print("执行操作 ... ")
    #解锁
    sem.release()


def singsong_ktv(person,sem):
    # 上锁
    sem.acquire()
    
    print("{}进入了唱吧ktv , 正在唱歌 ~".format(person))
    # 唱一段时间
    time.sleep( random.randrange(4,8) ) # 4 5 6 7
    
    print("{}离开了唱吧ktv , 唱完了 ... ".format(person))
    
    # 解锁
    sem.release()

if __name__ == "__main__":
    sem = Semaphore(5)
    lst = ["赵凤勇" , "沈思雨", "赵万里" , "张宇" , "假率先" , "孙杰龙" , "陈璐" , "王雨涵" , "杨元涛" , "刘一凤"   ]
    for i  in lst:
        p = Process(target=singsong_ktv , args = (i , sem)        )
        p.start()

#总结: Semaphore 可以设置上锁的数量 , 同一时间上多把锁

创建进程时,是异步并发,执行任务时,是同步程序;

创建几把Semaphore锁。同时就可以允许几个进程执行，当有进程执行完毕，其他进程才能获得锁，来执行，同时允许正在执行最大为设置锁的个数

其他的进程异步并发等待，直到执行时是同步执行

#赵万里进入了唱吧ktv , 正在唱歌 ~

#赵凤勇进入了唱吧ktv , 正在唱歌 ~

#张宇进入了唱吧ktv , 正在唱歌 ~

#沈思雨进入了唱吧ktv , 正在唱歌 ~

#孙杰龙进入了唱吧ktv , 正在唱歌 ~

同一时间，只能五个人唱歌，离开一个，下一个进程抢到锁，达到了多个进程上多把锁的效果

进程并不是开的越多越好，因为每开一个进程，都会开辟空间，占用系统资源。内存资源是有限的。内存CPU资源不足时，进程很多也会很卡顿

我们可以用Semaphore来限制开辟进程的数量。后面我们使用进程池来限制进程数量，使用锁来控制

进程队列

from multiprocessing import Process,Queue

#引入线程模块; 为了捕捉queue.Empty异常;

import queue

1.基本语法

顺序: 先进先出,后进后出

#创建进程队列

q = Queue()

#put() 存放

q.put(1)

q.put(2)

q.put(3)

#get() 获取

“”“在获取不到任何数据时,会出现阻塞”“”

#print( q.get() )

在获取不到任何数据时,会出现阻塞

#get_nowait() 拿不到数据报异常，不等待

Windows下，拿不到数据报异常

队列判断空与不空，实际上判断的是线程的队列空与不空，不是进程的

#引入线程模块; 为了捕捉queue.Empty异常;

[windows]效果正常 [linux]不兼容

linux下get_nowait()不能使用

Linux下，获取队列时，如果先用get()获取第一个队列元素，后面的可以用get_nowait()来获取，不再报错

不管什么平台，为了防止报错，我们可以抑制异常

try:
    print(  q.get_nowait()  )
    print(  q.get_nowait()  )
    print(  q.get_nowait()  )
    print(  q.get_nowait()  )
except : #queue.Empty
    pass


# put_nowait() 非阻塞版本的put
# 设置当前队列最大长度为3 ( 元素个数最多是3个 )
"""在指定队列长度的情况下,如果塞入过多的数据,会导致阻塞"""
# q2 = Queue(3)
# q2.put(111)
# q2.put(222)
# q2.put(333)
# q2.put(444)

使用put_nowait 在队列已满的情况下,塞入数据会直接报错

q2 = Queue(3)
try:
    q2.put_nowait(111)
    q2.put_nowait(222)
    q2.put_nowait(333)
    q2.put_nowait(444)
except:
    pass

判断队列是否塞满，用的也是线程的，不是进程的

linux也支持put_nowait()添加队列，塞满报错

2.进程间的通信IPC

#IPC Inter-Process Communication

#实现进程之间通信的两种机制:

# 管道 Pipe 现在管道基本不用了，没有Queue强大

# 队列 Queue

# put() 存放
# get() 获取
# get_nowait() 拿不到报异常
# put_nowait() 非阻塞版本的put
q.empty()      检测是否为空  (了解)
q.full()        检测是否已经存满 (了解)
q.qsize()     检测队列元素个数(了解)

这几个标记了解的不太好用，因为生产中但凡涉及到队列，都是多进程之间的。这一刻这个进程判断队列非空，下一秒另一个进程把数据拿走了，实时性不对，所以没有判断的必要

管道

匿名管道

管道是一种半双工的通信方式，数据只能单向流动，而且只能在具有亲缘关系的进程间使用。进程的亲缘关系通常是指父子进程关系。

命名管道

有名管道也是半双工的通信方式，但是它允许无亲缘关系进程间的通信。

消息队列

消息队列是由消息的链表，存放在内核中并由消息队列标识符标识。消息队列克服了信号传递信息少、管道只能承载无格式字节流以及缓冲区大小受限等缺点。

通过消息队列交换数据，这样能够极大地减少了对使用锁定和其他同步手段的需求

def func(q):
    # 2.子进程获取主进程存放的数据
    res = q.get()
    print(res,"<22>")
    # 3.子进程中存放数据
    q.put("刘一缝")

if __name__ == "__main__":
    q3 = Queue()
    p = Process(target=func,args=(q3,))
    p.start()

    # 1.主进程存入数据
    q3.put("赵凤勇")
    
    # 为了等待子进程把数据存放队列后,主进程在获取数据;
    p.join()
    
    # 4.主进程获取子进程存放的数据
    print(q3.get() , "<33>")

队列，塞入一个数据后，get / get_nowait一次后，就没了。不能多次获取同一个位置的值

第一步，主进程将王一博存入队列。第二步：子进程获取队列中的值，打印出来是王一博。并且向队列中插入数据景浩

第三步，使用进程同步，让子进程执行完毕再执行下面的代码，由于队列中第一个数据王一博已被子进程获取，所以此次主进程只获取到子进程插入的数据景浩

总结

对于共享内存，数据操作最快，因为是直接在内存层面操作，省去中间的拷贝工作。但是共享内存只能在单机上运行，且只能操作基础数据格式，无法直接共享复杂对象。

管道和队列传递数据没有共享内存快，且每次传递的数据大小受限。但是使用队列可以在多个进程间传递，可以在不同主机上的进程间共享，以实现分布式。

匿名管道则只能在父子进程间共享，而命名管道可在同一台计算机的不同进程之间或在跨越一个网络的不同计算机的进程间共享。

Python的进程锁，进程队列

锁 lock 互斥锁

模拟12306 抢票软件

信号量 Semaphore

进程队列

1.基本语法

2.进程间的通信IPC

热门文章

最新文章

相关课程

相关电子书

相关实验场景