12 ，生产者消费者模型 ，队列作为缓冲区 ，Queue ，JoinableQueue ，精彩的消费者生产者两段分析_生产者消费者问题缓冲队列

1 ，生产者消费者模型 ： 目的

解耦

2 ，生产者消费者模型 ： 图示

3 ，队列，栈 ：

1. 队列 ： 先进先出 ( FILO )
2. 栈 ： 先进后出 ( FIFO )

4 ，不同的队列 ：

1. queue ： 不可以多进程
2. from multiprocessing import Queue ： 可以多进程，并且数据安全，不需要加锁

5 ，生产者 - 消费者 ：

1. 目的 ： 用队列实现 ( 生产者-消费者 ) 模型，先生产，再消费
2. 思路 ：
   1 ，生产者 ： 线程 1 ，负责生产数据
   2 ，消费者 ： 线程 2 ，负责消费数据
   3 ，缓冲区 ： 多线程中的队列模块
3. 代码 ：

# Author:SFL
from multiprocessing import Process
from multiprocessing import Queue

# 生产者
def pro(q:Queue):
    for i in range(1000):
        q.put("对面谁上单 %i"%(i))
        print("对面谁上单 %i"%(i))
    # 放完了所有数据以后，放一个 None ，作为结束的标志
    q.put(None)

# 消费者
def con(q:Queue):
    i = 0
    while 1:
        r = q.get()
        if r:
            print("你爹 %i"%(i))
            i = i+1
        else :
            break
if __name__ == '__main__':
    # 缓冲区 ： 队列 ( 长度 5 )
    q = Queue(5)
    p1 = Process(target=pro, args=(q,))
    p2 = Process(target=con, args=(q,))
    p1.start()
    p2.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

4. 结果 ：

对面谁上单 0
对面谁上单 1
对面谁上单 2
对面谁上单 3
对面谁上单 4
你爹 0
对面谁上单 5
你爹 1
你爹 2
你爹 3
你爹 4
你爹 5
对面谁上单 6
对面谁上单 7
1
2
3
4
5
6
7
8
9
10
11
12
13
14

6 ，生产者 - 消费者 ： 进阶 1 ( 结束语写在父进程中 )

# Author:SFL
from multiprocessing import Process
from multiprocessing import Queue

# 生产者
def pro(q:Queue):
    for i in range(1000):
        q.put("对面谁上单 %i"%(i))
        print("对面谁上单 %i"%(i))

# 消费者
def con(q:Queue):
    i = 0
    while 1:
        r = q.get()
        if r:
            print("你爹 %i"%(i))
            i = i+1
        else :
            break
if __name__ == '__main__':
    # 缓冲区 ： 队列 ( 长度 5 )
    q = Queue(5)
    p1 = Process(target=pro, args=(q,))
    p2 = Process(target=con, args=(q,))
    p1.start()
    p2.start()
    # 让 p1 进程，优先于主进程
    p1.join()
    q.put(None)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

7 ，多个生产者，多个消费者 ： JoinableQueue ( 理论 )

1. 作用 ： 用于多进程之间的通讯
2. tesk_done ： 告诉消息队列，我 get 好了，可以下一条消息了
3. join ： 等着消费者，将所有消息都消费掉，再停止运行

8 ，3 个生产者，2 个消费者 ： JoinableQueue ( 代码 )

# Author:SFL
from multiprocessing import Process
from multiprocessing import Queue
from multiprocessing import JoinableQueue

# 生产者
def pro(q:JoinableQueue):
    for i in range(1000):
        q.put(i)
        print("对面谁上单 %i"%(i))
    # 等着其他进程，将队列中的消息都消费掉
    q.join()

# 消费者
def con(q:JoinableQueue):
    while 1:
        r = q.get()
        print("你爹 %i"%(r))
        # 本次消息消费结束
        q.task_done()

if __name__ == '__main__':
    # 缓冲区 ： 队列 ( 长度 5 )
    q = JoinableQueue(5)
    p1 = Process(target=pro, args=(q,))
    p2 = Process(target=pro, args=(q,))
    c1 = Process(target=con, args=(q,))
    c2 = Process(target=con, args=(q,))
    c3 = Process(target=con, args=(q,))
    c1.daemon = True
    c2.daemon = True
    c3.daemon = True
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    c3.start()
    # 生产者必须在主进程之前执行，主进程不可以提前结束
    p1.join()
    p2.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

9 ，对于 2 生 3 销代码的精彩分析 ：

1. 主程序等待生产者进程结束 ( main < p1,p2 )
2. 一共 6 个进程 ： 1 主，2 生，3 消
3. 生产者进程会等待消费者，将所有消息都消费掉，才结束 ( p1,p2 < c1,c2,c3 )
4. 出现一个问题 ： 消费者在 while 1 中，永远的死循环，进程不能结束
5. 解决 ：
   1 ，生产者，会在队列中完全没有消息的时候结束
   2 ，消费者，只有消费者会永远不结束
   3 ，将消费者设置为守护进程
   4 ，好处 ： 当生产者进程结束，消费者就强制结束，不管你还有没有事情做
   5 ，不会发生数据错乱吗 ：不会
   6 ，技巧 ： 发生在生产者和消费者身上