#并发运行,效率高,但竞争同一打印终端,带来了打印错乱from multiprocessing import Processimport os,timedef work():print('%s is running' %os.getpid())time.sleep(2)print('%s is done' %os.getpid())​if __name__ == '__main__':for i in range(3):p=Process(target=work)p.start()

  #加锁后由并发变成了串行,牺牲了运行效率,但避免了竞争​from multiprocessing import Process,Lockimport os,timedef work(mutex):mutex.acquire() #开始加锁print('%s is running' %os.getpid())time.sleep(2)print('%s is done' %os.getpid())mutex.release() #释放锁，在加锁期间别的进程都要等​if __name__ == '__main__':mutex = Lock()for i in range(3):p=Process(target=work,args=(mutex,))p.start()

  #文件db.txt的内容为：{"count":1}#注意一定要用双引号，不然json无法识别​# 并发运行，效率高，但竞争写同一文件，数据写入错乱from multiprocessing import Process,Lockimport time,json,random,osdef search():dic=json.load(open('db.txt'))print('\033[43m剩余票数%s\033[0m' %dic['count'])​def get():dic=json.load(open('db.txt'))time.sleep(0.1) #模拟读数据的网络延迟if dic['count'] >0:dic['count']-=1time.sleep(0.2) #模拟写数据的网络延迟json.dump(dic,open('db.txt','w'))print('%s\033[43m购票成功\033[0m'%(os.getpid()))​def task(lock):search()get()if __name__ == '__main__':lock=Lock()for i in range(10): #模拟并发10个客户端抢票p=Process(target=task,args=(lock,))p.start()

  剩余票数1剩余票数1剩余票数1剩余票数1剩余票数1剩余票数1剩余票数1剩余票数1剩余票数1剩余票数14120购票成功2692购票成功7328购票成功13444购票成功13632购票成功13560购票成功13752购票成功12564购票成功13720购票成功13488购票成功

  import multiprocessing,time,json,random​def search(name):with open("db.txt","r",encoding="utf-8") as f:data_dic = json.load(f)time.sleep(random.uniform(0,2))if data_dic["count"] >= 1 :print("已查询到票还有%s张,当前系统时间 %s"%(data_dic["count"],time.asctime()))else:print("系统票源不足！当前系统时间 %s"%time.asctime())​def buy(name):with open("db.txt","r+",encoding="utf-8") as f:data_dic = json.load(f)if data_dic["count"] > 0 :with open("db.txt", "w", encoding="utf-8") as g:new_ticket_count = data_dic["count"] - 1data_dic.update({"count":new_ticket_count})json.dump(data_dic,g)print("%s购票成功！"%name)else:print("%s购票失败！"%name)​def task(name,mutex):search(name)  # 查询无需加锁mutex.acquire()buy(name)  #针对修改文件的关键操作加锁mutex.release()​​if __name__ == "__main__":mutex = multiprocessing.Lock()for i in range(10):p = multiprocessing.Process(target=task,args=("乘客%s"%i,mutex))p.start()

  #加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。虽然可以用文件共享数据实现进程间通信，但问题是：1.效率低（共享数据基于文件，而文件是硬盘上的数据）2.需要自己加锁处理​​​#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。1 队列和管道都是将数据存放于内存中2 队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

  # 多进程中的组件# ktv# 4个# 一套资源  同一时间 只能被n个人访问# 某一段代码 同一时间 只能被n个进程执行import timeimport randomfrom multiprocessing import Processfrom multiprocessing import Semaphore​# sem = Semaphore(4)# sem.acquire()# print('拿到第一把钥匙')# sem.acquire()# print('拿到第二把钥匙')# sem.acquire()# print('拿到第三把钥匙')# sem.acquire()# print('拿到第四把钥匙')# sem.acquire()# print('拿到第五把钥匙')def ktv(i,sem):sem.acquire()    #获取钥匙print('%s走进ktv'%i)time.sleep(random.randint(1,5))print('%s走出ktv'%i)sem.release()​​if __name__ == '__main__' :sem = Semaphore(4)for i in range(20):p = Process(target=ktv,args=(i,sem))p.start()

  from multiprocessing import Event​e = Event()print(e.is_set()) #初始设置为Falseprint("数据111")e.set()  #设置之后为Trueprint("数据222")print(e.is_set()) #打印设置之后的状态e.wait()  #当值为False会阻塞，当值为Ture是，不会阻塞print("数据333")e.clear() #清除事件状态，设置为Falseprint(e.is_set())  #打印清除之后的状态print("数据444")e.wait()  #此时值为False，程序会一直阻塞print("数据555")

  False数据111数据222True数据333False数据444

  from multiprocessing import Event,Processimport timeimport random​​def cars(e,num):if not e.is_set(): # 进程刚开启，is_set()的值是False，模拟信号灯为红色print("%s车正在等待通行"%num)e.wait() # 阻塞，等待信号灯切换print("%s车已经通过" % num) #打印已经通过的进程​​def light(e):​#模拟定时切换红绿灯while True:if e.is_set():e.clear() #>将is_set()的值设置为Falseprint("\033[31m红灯亮了\033[0m")else:e.set() #>将is_set()的值设置为Trueprint("\033[32m绿灯亮了\033[0m")time.sleep(2)​if __name__ == "__main__":e = Event()traffic = Process(target=light,args=(e,))traffic.start() #启动红绿灯进程for i in range(20):car = Process(target=cars,args=(e,"布加迪%s"%i))car.start()time.sleep(random.random())

  Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

  maxsize是队列中允许最大项数，省略则无大小限制。

  q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.q.get_nowait():同q.get(False)q.put_nowait():同q.put(False)​q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

  q.close() 关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。​q.cancel_join_thread() 不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。​q.join_thread() 连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

  from multiprocessing import Queue​q = Queue(3)  # 创建一个队列对象，并给他设置容器大小，即能放几个数据q.put(1)  # put（）方法是往容器里放数据q.put([2,3])q.put({"k1":4})# q.put("mi") # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。try:q.put_nowait(3) # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。print('队列已经满了')​# 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。print(q.full()) #返回True ,满了print(q.get())  #get（）方法是从容器里拿数据print(q.get())print(q.get())# 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。try:q.get_nowait() # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。print('队列已经空了')​print(q.empty()) #空了

  import timefrom multiprocessing import Queue, Process​​def task(q):q.put(" hello! 时间%s"%time.asctime())  # 调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。​​if __name__ == '__main__':q = Queue(3)#创建一个Queue对象p = Process(target=task, args=(q,)) #创建一个子进程p.start()print(q.get()) #在主进程打印从子进程获取的数据

  from multiprocessing import Process, Queueimport time, random, os​​def consumer(q):while True:res = q.get()time.sleep(random.randint(1, 3))print('\033[45m%s 吃 %s\033[0m' % (os.getpid(), res))​​def producer(q):for i in range(10):time.sleep(random.randint(1, 3))res = '包子%s' % iq.put(res)print('\033[44m%s 生产了 %s\033[0m' % (os.getpid(), res))​​if __name__ == '__main__':q = Queue()# 生产者们:即厨师们p1 = Process(target=producer, args=(q,))​# 消费者们:即吃货们c1 = Process(target=consumer, args=(q,))​# 开始p1.start()c1.start()print('主')

  #程序中有两类角色一类负责生产数据（生产者）一类负责处理数据（消费者）#引入生产者消费者模型为了解决的问题是：平衡生产者与消费者之间的工作能力，从而提高程序整体处理数据的速度#如何实现：生产者<-->队列<——>消费者#生产者消费者模型实现类程序的解耦和

  import time, random, osfrom multiprocessing import Process, Queue​​def consumer(q):while True:res = q.get()if res is None: break  # 收到结束信号则结束time.sleep(random.randint(1, 3))print('\033[45m%s 吃 %s\033[0m' % (os.getpid(), res))​​def producer(q):for i in range(10):time.sleep(random.randint(1, 3))res = '包子%s' % iq.put(res)print('\033[44m%s 生产了 %s\033[0m' % (os.getpid(), res))q.put(None)  # 发送结束信号，生产者在生产完毕后发送结束信号None​​if __name__ == '__main__':q = Queue()# 生产者们:即厨师们p1 = Process(target=producer, args=(q,))​# 消费者们:即吃货们c1 = Process(target=consumer, args=(q,))​# 开始p1.start()c1.start()print('主')

  import multiprocessingimport timeimport random​​def producer(name, q):for i in range(2):res = "包子%s" % itime.sleep(random.randint(0, 1))print("%s生产了%s" % (name, res))q.put(res)​​def consumer(name, q):while True:res = q.get()if q.get() is None:  # 收到结束信号则结束print("没包子吃了")breakprint("%s吃了%s" % (name, res))​​if __name__ == "__main__":q = multiprocessing.Queue()p1 = multiprocessing.Process(target=producer, args=("jack", q))p2 = multiprocessing.Process(target=producer, args=("charles", q))p3 = multiprocessing.Process(target=producer, args=("pony", q))c1 = multiprocessing.Process(target=consumer, args=("nick", q))c2 = multiprocessing.Process(target=consumer, args=("nicholas", q))p_list = []p_list.append(p1)p_list.append(p2)p_list.append(p3)for p in p_list:p.start()c1.start()c2.start()p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号p2.join()p3.join()q.put(None)  # 发送结束信号，有几个消费者就应该发送几次结束信号Noneq.put(None)  # 发送结束信号print("end........")

  JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：​q.task_done() 使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。​q.join() 生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task_done()方法为止。 下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

import multiprocessing
import time
import randomdef producer(name, q):for i in range(2):res = "包子%s" % itime.sleep(random.randint(0, 1))print("%s生产了%s" % (name, res))q.put(res)q.join()  # 只有顾客把队列的包子全部拿走后，三个生产者进程才能全部结束def consumer(name, q):while True:res = q.get()print("%s吃了%s" % (name, res))q.task_done()  # 发信号告诉队列，又吃完了一个，从队列中取走一个数据并处理完成if __name__ == "__main__":# q = multiprocessing.Queue()q = multiprocessing.JoinableQueue()p1 = multiprocessing.Process(target=producer, args=("jack", q))p2 = multiprocessing.Process(target=producer, args=("charles", q))p3 = multiprocessing.Process(target=producer, args=("pony", q))c1 = multiprocessing.Process(target=consumer, args=("nick", q))c2 = multiprocessing.Process(target=consumer, args=("nicholas", q))p_list = []p_list.append(p1)p_list.append(p2)p_list.append(p3)for p in p_list:p.start()c1.daemon = True  # 将c1\c2设置成守护进程，只要主进程结束了，那么顾客就收到了所有的数据c2.daemon = Truec1.start()c2.start()p1.join()p2.join()p3.join()print("end........")
# 主进程等--->p1,p2,p3等---->c1,c2
# p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据
# 因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。
# 应该随着主进程的结束而结束,所以设置成守护进程就可以了。