python 并发编程 锁 / 信号量 / 事件 / 队列(进程间通信（IPC）) /生产者消费者模式

Posted 2021-01-14 qq1426794755

(1)锁:进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的，而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理。

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

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

def get_ticket(i,ticket_lock):
    print(‘我们到齐了‘)
    time.sleep(1)
    ticket_lock.acquire()
    with open(‘text‘, ‘r‘) as f:
        last_ticket_info = json.load(f)
    last_ticket = last_ticket_info[‘count‘]
    if last_ticket > 0:
        time.sleep(random.random())
        last_ticket = last_ticket - 1
        last_ticket_info[‘count‘] = last_ticket
        with open(‘text‘,‘w‘) as f:
            json.dump(last_ticket_info,f)
        print(‘%s号抢到了。‘%i)
    else:
        print("%s号傻逼，没抢到"%i)
    ticket_lock.release()

if __name__ == ‘__main__‘:
    ticket_lock = Lock()
    for i in range(10):
        p = Process(target=get_ticket,args=(i,ticket_lock,))
        p.start()


##############
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
我们到齐了
4号傻逼，没抢到
5号傻逼，没抢到
0号傻逼，没抢到
2号傻逼，没抢到
1号傻逼，没抢到
3号傻逼，没抢到
8号傻逼，没抢到
7号傻逼，没抢到
6号傻逼，没抢到
9号傻逼，没抢到

抢票示例

(2)信号量:　　　　(Semaphore)

from multiprocessing import Process,Semaphore    ##  信号量  ##
import time

def dabaojian(i,s):
    s.acquire()
    print(‘%s号来洗脚‘%i)
    time.sleep(1)
    s.release()

if __name__ == ‘__main__‘:
    s = Semaphore(4)
    for i in range(10):
        p = Process(target=dabaojian,args=(i,s,))
        p.start()

#########
5号来洗脚
0号来洗脚
1号来洗脚
4号来洗脚

2号来洗脚
3号来洗脚
7号来洗脚
6号来洗脚

8号来洗脚
9号来洗脚

Process finished with exit code 0

洗脚示例

(3)事件: 事件完成红绿灯示例　　　　(Event)

from multiprocessing import Process,Event     ####  事件  ####
import time


def taffic_lights(e):
    while 1:
        print(‘红灯亮‘)
        time.sleep(5)

        e.set()
        print(‘绿灯亮‘)
        time.sleep(3)
        e.clear()

def car(i,e):

        if not e.is_set():
            print(‘我们在等待。。。‘)
            e.wait()
            print(‘走你‘)
        else:
            print(‘可以走了。。‘)


if __name__ ==‘__main__‘:
    e = Event()
    hld = Process(target=taffic_lights,args=(e,))
    hld.start()

    while 1:
        time.sleep(0.5)

        for i in range(3):
            p = Process(target=car,args=(i,e,))
            p.start()

#################################
红灯亮
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
我们在等待。。。
绿灯亮
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
走你
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
可以走了。。
红灯亮

事件模拟完成红绿灯示例

(4)队列:进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。队列就像一个特殊的列表，但是可以设置固定长度，并且从前面插入数据，从后面取出数据，先进先出。

Queue([maxsize]) 创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。
底层队列使用管道和锁实现。

queue的方法介绍:

q = Queue([maxsize]) 
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。 
Queue的实例q具有以下方法：

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

q.get_nowait( ) 
同q.get(False)方法。

q.put(item [, block [,timeout ] ] ) 
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

q.qsize() 
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。


q.empty() 
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

q.full() 
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。

代码示例:

from multiprocessing import Process,Queue         #####  队列  #####

q = Queue(3)

q.put(1)
print(q.full())
q.put(2)
q.put(3)
print(q.full())

print(q.get())
print(q.empty())
print(q.get())
print(q.get())
print(q.empty())
print(q.get())

while 1:
    try:
        q.get(False)
        # q.get_nowait()
    except:
        print(‘队列目前为空‘)

##########################
False
True
1
False
2
3
True

队列的简单示例

 

子进程和父进程通过队列进行通信

from multiprocessing import Process,Queue
import time

def girl(q):
    print(‘来自boy的信息‘,q.get())
    print(‘来自领导的凝视‘,q.get())

def boy(q):
    q.put(‘约么？‘)


if __name__ == ‘__main__‘:
    q = Queue(5)
    boy_p = Process(target=boy,args=(q,))
    girl_p = Process(target=girl,args=(q,))
    boy_p.start()
    girl_p.start()
    time.sleep(1)
    q.put(‘好好工作，别乱搞‘)

##########################
来自boy的信息 约么？
来自领导的凝视 好好工作，别乱搞

子进程与父进程通过队列进行通信