JoinableQueue同样通过multiprocessing使用。

创建队列的另外一个类：

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

参数介绍：

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

方法介绍：

    JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

    q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

    q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

1 import time
2 import random
3 from multiprocessing import Process,JoinableQueue
4
5 def consumer(name,q):
6 while True:
7 res=q.get()
8 if res is None:break
9 time.sleep(random.randint(1,3))
10 print('\033[46m消费者===》%s 吃了 %s\033[0m' %(name,res))
11 q.task_done()
12
13 def producer(name,q,food):
14 for i in range(5):
15 time.sleep(random.randint(1,2))
16 res='%s%s' %(food,i)
17 q.put(res)
18 print('\033[45m生产者者===》%s 生产了 %s\033[0m' %(name,res))
19
20
21
22 if __name__ == '__main__':
23 #1、共享的盆
24 q=JoinableQueue()
25
26
27 #2、生产者们
28 p1=Process(target=producer,args=('egon',q,'包子'))
29 p2=Process(target=producer,args=('刘',q,'泔水'))
30 p3=Process(target=producer,args=('杨',q,'米饭'))
31
32 #3、消费者们
33 c1=Process(target=consumer,args=('alex',q))
34 c2=Process(target=consumer,args=('梁',q))
35 c1.daemon=True
36 c2.daemon=True
37
38 p1.start()
39 p2.start()
40 p3.start()
41 c1.start()
42 c2.start()
43
44
45 # 确定生产者确确实实已经生产完毕
46 p1.join()
47 p2.join()
48 p3.join()
49 # 在生产者生产完毕后，拿到队列中元素的总个数，然后直到元素总数变为0，q.join()这一行代码才算运行完毕
50 q.join()
51 #q.join()一旦结束就意味着队列确实被取空，消费者已经确确实实把数据都取干净了
52 print('主进程结束')

生产者消费者模型例子

1、什么是线程

线程指的是一条流水线的工作过程

进程根本就不是一个执行单位，进程其实是一个资源单位
一个进程内自带一个线程，线程才是执行单位

2、进程VS线程

• 同一进程内的线程们共享该进程内资源，不同进程内的线程资源肯定是隔离的

• 创建线程的开销比创建进程要小的多

• 一个进程的所有线程的pid都是一样的

• 线程之间的地位相同，没有主次之分，但是主线程又代表着主进程

  　　3、创建线程的两种方式　　　　

#方式一
from threading import Thread
import time
def sayhi(name):
print('%s say hello' %name)

if __name__ == '__main__':
t=Thread(target=sayhi,args=('egon',))
t.start()
print('主线程')

#方式二
from threading import Thread
import time
class Sayhi(Thread):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
print('%s say hello' % self.name)

if __name__ == '__main__':
t = Sayhi('egon')
t.start()
print('主线程')

　　　　线程执行的结果：

　　　　　　

　　　　与进程不同，是因为线程启动的代价很小，启动速度远大于进程，所以，子线程先打印，主线程才打印

　　4、线程的其他函数　　　　　　

from threading import Thread,current_thread,active_count,enumerate
import time,os

def task():
print('%s is running' %current_thread().name)
time.sleep(3)

if __name__ == '__main__':
t1=Thread(target=task,name='第一个线程')
t2=Thread(target=task,)
t3=Thread(target=task,)
t1.start()
t2.start()
t3.start()

# print(t1.is\_alive())  #查看线程是否存活  
print(active\_count())  #查看活跃的线程数，包括主线程  
print(enumerate())     #获取当前每个活跃的线程对象  
print('主线程',current\_thread().name)   #获取当前线程名

Thread实例对象的方法
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。

threading模块提供的一些方法：
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

5、守护线程

t.daemon=true

设为守护线程，守护线程要在线程开始执行之前设置，否则报错

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

需要强调的是：运行完毕并非终止运行

1 .对主进程来说，运行完毕指的是主进程代码运行完毕

2. 对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

　   详细解释：　　　　

from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)

if __name__ == '__main__':
t=Thread(target=sayhi,args=('egon',))
t.setDaemon(True) #必须在t.start()之前设置
t.start()

print('主线程')  
print(t.is\_alive())  
'''  
主线程  
True  
'''

死锁现象

　　　　所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

1 from threading import Thread,Lock,RLock
2 import time
3
4 mutexA=Lock()
5 mutexB=Lock()
6
7 class MyThread(Thread):
8 def run(self):
9 self.f1()
10 self.f2()
11
12 def f1(self):
13 mutexA.acquire()
14 print('%s 拿到了A锁' %self.name)
15
16 mutexB.acquire()
17 print('%s 拿到了B锁' %self.name)
18 mutexB.release()
19
20 mutexA.release()
21
22 def f2(self):
23 mutexB.acquire()
24 print('%s 拿到了B锁' %self.name)
25 time.sleep(0.1)
26
27 mutexA.acquire()
28 print('%s 拿到了A锁' %self.name)
29 mutexA.release()
30
31 mutexB.release()
32
33
34 if __name__ == '__main__':
35 for i in range(10):
36 t=MyThread()
37 t.start()
38
39 print('主')

死锁

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

　 信号量

同进程的一样

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)： 　　

# from multiprocessing import Semaphore

from threading import Thread,Semaphore,current_thread
import time,random

sm=Semaphore(5)

def go_wc():
sm.acquire()
print('%s 上厕所ing' %current_thread().getName())
time.sleep(random.randint(1,3))
sm.release()

if __name__ == '__main__':
for i in range(23):
t=Thread(target=go_wc)
t.start()