#文件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']-=1time.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']-=1time.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()使用锁来保证数据安全

使用锁来保证数据安全

from multiprocessing import Process,Semaphore
import time,randomdef 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('============》')例子

from multiprocessing import Process, Event
import time, randomdef 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语句开始continueprint('车开远了,car', n)breakdef 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)breakdef traffic_lights(e, inverval):while True:time.sleep(inverval)if e.is_set():print('######', e.is_set())e.clear()  # ---->将is_set()的值设置为Falseelse:e.set()    # ---->将is_set()的值设置为Trueprint('***********',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('============》')红绿灯实例