锁,信号量和事件,都是用来同步异步程序的组件.

锁

from multiprocessing import Process
import os, time, json


def show(i):
    with open('ticket') as f:
        dic = json.load(f)
        print('编号{}查看了余票: %s'.format(i,dic['ticket']))

def buy(i):
    with open('ticket',) as f:
        dic = json.load(f)
        if dic['ticket'] >0:
            dic['ticket'] -= 1
            print('{}买到票了'.format(i))
        else:
            print('{}没买到票'.format(i))
    time.sleep(0.1)
    with open('ticket','w') as f:
        json.dump(dic,f)

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

这个程序想要实现查看文件里的剩余票数,如果有票,就将票数减少1,如果为0,就返回买不到票.
但是实际运行起来,发现10个人都买到了票,这显然有问题,是因为进程运行都很快,在第一个访问文件的进程将0写入文件之前,其他进程也读取了1,结果都拿到一样的结果.
如果要改变这种情况,就在文件的门口放一把锁和挂上唯一的一把钥匙.先拿到钥匙的人把门锁上,完成工作之后,打开门然后把钥匙挂在门上,下一个人来的时候,看到门锁着,则无法访问文件,看到有钥匙,则开门进房间.
这个过程抽象成一个锁的概念.锁的使用方法:在需要使用锁的程序代码前后,用Lock()实例化的对象的acquire方法来获得锁,release方法来释放锁.

import json
import time
from multiprocessing import Process
from multiprocessing import Lock


def buy_ticket(i, lock):
    lock.acquire()  # 拿钥匙进门
    with open('ticket') as f:
        dic = json.load(f)
        time.sleep(0.1)
    if dic['ticket'] > 0:
        dic['ticket'] -= 1
        print('\033[32m%s买到票了\033[0m' % i)
    else:
        print('\033[31m%s没买到票\033[0m' % i)
    time.sleep(0.1)
    with open('ticket', 'w') as f:
        json.dump(dic, f)
    lock.release()  # 还钥匙


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

    # 以下是恢复票数
    time.sleep(3)
    with open('ticket') as f:
        dic = json.load(f)
        dic['ticket'] = 3
    with open('ticket','w') as f:
        json.dump(dic, f)

这样每次可以看到,10个人里确实有3个人抢到了票.
注意,启动的10个进程,传入的参数是同一个对象.

信号量

在锁的基础上,如果同时有n个资源可以供超过这些资源的数目进程访问,那么同时只能让n个进程分别访问这个资源,就相当于指定数量的锁,这个时候就用信号量类,来实例化一个信号量对象,用于控制总数.

import time
import random
from multiprocessing import Process
from multiprocessing import Semaphore


def ktv(i, sem):
    sem.acquire()   # 取一个钥匙然后进门
    print('{}进入了KTV'.format(i))
    time.sleep(random.randint(2, 5))
    print('{}离开了KTV'.format(i))
    sem.release()   # 归还钥匙


if __name__ == '__main__':
    sem = Semaphore(4)
    for i in range(10):
        p = Process(target=ktv, args=(i, sem))
        p.start()

这个程序模拟了一个只能容纳4个人的KTV房间,有10个人想唱歌,则必须等进去的人唱好出来之后,才能再进去.
实际上把sem.acquire()直接写在主进程里边,可以看到,反复调用sem.acquire(),只有前四次,之后的代码会响应,之后就会阻塞住.其实就相当于一个加了计数器的锁.

事件

对于锁和信号量可以使所有的进程都进入阻塞状态,也可以控制所有的进程解除阻塞.
一个事件被创建之后,默认是阻塞状态.

from multiprocessing import Event
if __name__ == '__main__':
    e = Event()  # 实例化事件
    print(e.is_set())  # 查看一个事件的状态,默认被设置成阻塞
    print(123456)

结果是False,并且打印了123456,似乎没有阻塞,实际上,事件e是根据e.wait()的值来决定是否阻塞.

if __name__ == '__main__':
    e = Event()
    print(e.is_set())
    e.set()  # 设置事件的状态为True
    print(e.is_set())  # 用来查看一个事件的状态
    e.wait() # 是否阻塞取决于set是否为True
    print(123456)
    e.clear()  # 将事件的状态重新设置为False
    e.wait()
    print(123456)

第一个123456正常打印,第二个123456之前事件e发生阻塞.事件被创建的时候默认为阻塞,并且会在wait()方法的地方阻塞.设置为True的时候,不阻塞.

事件的作用,就是等待某一件事中的一个信号变化,然后做一些动作,再将状态改回来,继续再进入等待状态.这几个进程间同步控制,内部其实写了基于文件的socket,所以如果用错了,可能会出现socket错误.
用一个经典例子解释事件:

import time
from multiprocessing import Process
from multiprocessing import Event


def light(e):
    while True:
        if e.is_set():
            time.sleep(2)
            e.clear()
            print('\033[31m红灯亮了\033[0m ')

        else:
            time.sleep(2)
            e.set()
            print('\033[32m绿灯亮了\033[0m ')


def car(i, e):
    while True:
        if not e.is_set():
            print('{}车在等待通行')
            e.wait()
            print('{}车通过绿灯')


if __name__ == '__main__':
    e = Event()
    traffic = Process(target=light, args=(e,))
    traffic.start()

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

用一个事件传递到所有的进程中,这个事件的状态一旦改变,各个进程便可以通过这个事件的状态改变作出相应的逻辑.