#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcongfrom multiprocessing import Process,Lock
import timedef task(name):print('%s 1'%name)time.sleep(1)print('%s 2'%name)time.sleep(1)print('%s 3'%name)if __name__ == '__main__':for n in range(1,3):p = Process(target=task,args=('进程%d'%n,))p.start()'''输出：
进程1 1
进程2 1
进程1 2
进程2 2
进程1 3
进程2 3
'''

# Lock 互斥锁保证了进程间的有序运行，但效率降低
def task(name,mutex):mutex.acquire() # 加锁（添加互斥锁）print('%s 1'%name)time.sleep(1)print('%s 2'%name)time.sleep(1)print('%s 3'%name)mutex.release()  # 释放互斥锁，再运行其他进程if __name__ == '__main__':mutex = Lock()   # 父进程实例化一个互斥锁，子进程都会拷贝一份，需要传递给子进程for i in range(1,3):p = Process(target=task,args=('进程%d'%i,mutex))p.start()
''' 输出：
进程1 1
进程1 2
进程1 3
进程2 1
进程2 2
进程2 3
'''

#-*- coding:utf-8 -*-
# write by congcong
import json
import time
from multiprocessing import Process,Lockdef search(name):time.sleep(1)dic = json.load(open('data.json','r',encoding='utf-8'))print('<%s>查看剩余票数 【%d】'%(name,dic['count']))def buy(name):time.sleep(1)dic = json.load(open('data.json','r',encoding='utf-8'))if dic['count'] > 0:dic['count'] -= 1time.sleep(2)json.dump(dic,open('data.json','w',encoding='utf-8'))print('<%s> 购票成功！'%name)def task(name,mute):search(name) # 查询不需要加锁，可以并发执行mute.acquire() # 加锁，获得锁的进程才能继续运行，串行执行
    buy(name)mute.release() # 解锁if __name__ == '__main__':mute = Lock()for i in range(1,4):p = Process(target=task,args=('路人%d'%i,mute))p.start()
'''未加互斥锁时输出：
<路人2>查看剩余票数 【2】
<路人1>查看剩余票数 【2】
<路人3>查看剩余票数 【2】
<路人2> 购票成功！
<路人1> 购票成功！
<路人3> 购票成功！'''
'''加互斥锁后输出：
<路人1>查看剩余票数 【2】
<路人2>查看剩余票数 【2】
<路人3>查看剩余票数 【2】
<路人1> 购票成功！
<路人2> 购票成功！
'''

View Code

{"count": 2}

View Code

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong
import time
import json
from multiprocessing import Processdef search(name):time.sleep(1)dic = json.load(open('data.json','r',encoding='utf-8'))print('<%s> 剩余票数 [%d]'%(name,dic['count']))def buy(name):time.sleep(1)dic = json.load(open('data.json','r',encoding='utf-8'))if dic['count'] > 0:dic['count'] -= 1json.dump(dic,open('data.json','w',encoding='utf-8'))time.sleep(1)print('<%s> 购票成功！'%name)else:print('<%s> 购票失败！'%name)def task(name):search(name)buy(name)if __name__ == '__main__':for i in range(1,4):p = Process(target=task,args=('顾客%d'%i,))p.start()p.join()
'''
<顾客1> 剩余票数 [2]
<顾客1> 购票成功！
<顾客2> 剩余票数 [1]
<顾客2> 购票成功！
<顾客3> 剩余票数 [0]
<顾客3> 购票失败！
'''
# join 和 互斥锁 都可以将并发执行变为串行执行
'''
区别：join 对应修改的是全局的，而互斥锁则相对更加灵活，可以具体到全局的某一具体功能（共享）'''

View Code

{"count":2}

View Code

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

maxsize是队列中允许最大项数，省略则无大小限制。
但需要明确：1、队列内存放的是消息而非大数据2、队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

q.put 方法用以插入数据到队列中。
q.get 方法可以从队列读取并且删除一个元素。

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcongfrom multiprocessing import Queueq = Queue(3) # 括号内可指定最大存放数目，当未指定数目时，表示无限，但不建议这样，因为内存是有限的
q.put('hello world') # put()表示存放队列中（队列创建在内存中，可共享）
q.put({'a':1,'b':2})
q.put([1,2,3,4])
print(q.full())  #  True ---> full()判断队列是否已满
#q.put(666)  # 达到指定存放数目时，就会卡住，不再存放
print(q.get()) # hello world  ---> 队列为先进先出的线性表
print(q.get()) # {'a': 1, 'b': 2}
print(q.get()) # [1, 2, 3, 4]
print(q.empty()) # True   ---> empty() 判断队列是否为空
print(q.get()) # 当队列数据都完全取出后，再取不再运行

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong
import time
from multiprocessing import Process,Queue
# 解耦合；平衡了生产者和消费者的速度差
def producer(q):for i in range(1,6):good = '烤鸭%d'%itime.sleep(0.5)print('生产者生产了%s'%good)q.put(good)def consumer(q):while True:good = q.get()if good is None:breaktime.sleep(1)print('消费者吃了%s'%good)if __name__ == '__main__':# 容器q = Queue()# 生产者们p1 = Process(target=producer,args=(q,))p2 = Process(target=producer,args=(q,))p3 = Process(target=producer,args=(q,))# 消费者们c1 = Process(target=consumer,args=(q,))c2 = Process(target=consumer,args=(q,))c1.daemon = True # 守护进程，随主进程一同死去c2.daemon = Truep1.start()p2.start()p3.start()c1.start()c2.start()p1.join()p2.join()p3.join()q.put(None) # 结束信号，位于生产者所有的产品最后，栈尾q.put(None) # 结束信号的个数与消费者数量对应print('来客人了！') # 主进程
'''
1、主进程需要等生产者p1、p2、p3进程结束后运行
2、p1、p2、p3是在消费者把所有数据取完后才会结束
3、等p1、p2、p3结束了，消费者也就没有存在的意义了，应该随主进程一同死掉，故消费者应设成守护进程
'''

View Code

这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

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

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong
import time
from multiprocessing import Process,JoinableQueuedef producer(q):for i in range(1,5):good = '烤红薯%s'%itime.sleep(1)print('生产者生产了烤红薯%s'%good)q.put(good)q.join() # 等待队列进程执行完，即队列内数据被取完def consumer(q):while True:good = q.get()if good is None:breaktime.sleep(1)print('消费者吃了%s'%good)q.task_done() # 给生产者发送结束信号，即队列已经有一个数据被取走了if __name__ == '__main__':# 容器q = JoinableQueue()# 生产者p1 = Process(target=producer,args=(q,))p2 = Process(target=producer,args=(q,))p3 = Process(target=producer,args=(q,))# 消费者c1 = Process(target=consumer,args=(q,))c2 = Process(target=consumer,args=(q,))c1.daemon = True # 将消费者进程设为守护进程，以便消费者进程随主进程结束而结束c2.daemon = True# 发送信号
    p1.start()p2.start()p3.start()c1.start()c2.start()p1.join()p2.join()p3.join()print('主进程')

　　一类负责生产数据（生产者）一类负责处理数据（消费者）

平衡生产者与消费者之间的速度差
程序解开耦合

生产者<--->队列<--->消费者