进程

什么是进程：正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。但凡是硬件，都需要有操作系统去管理，只要有操作系统，就有进程的概念，就需要有创建进程的方式，一些操作系统只为一个应用程序设计，比如微波炉中的控制器，一旦启动微波炉，所有的进程都已经存在。

而对于通用系统（跑很多应用程序），需要有系统运行过程中创建或撤销进程的能力，主要分为4中形式创建新的进程

1. 系统初始化（查看进程linux中用ps命令，windows中用任务管理器，前台进程负责与用户交互，后台运行的进程与用户无关，运行在后台并且只在需要时才唤醒的进程，称为守护进程，如电子邮件、web页面、新闻、打印）

2. 一个进程在运行过程中开启了子进程（如nginx开启多进程，os.fork,subprocess.Popen等）

3. 用户的交互式请求，而创建一个新进程（如用户双击暴风影音）

4. 一个批处理作业的初始化（只在大型机的批处理系统中应用）

#!/urs/bin/evn python
# -*- coding:utf-8 -*-
import time
import random
from multiprocessing import Process


def main(name):
    print('%s names' % name)
    time.sleep(random.randrange(1, 5))
    print('%s names' % name)


if __name__ == '__main__':
    p1 = Process(target=main, args=('cc',))
    p2 = Process(target=main, args=('zq',))
    p1.start()
    p2.start()
    print('主')

注意：在windows中Process()必须放到# if __name__ == '__main__':下

#  创建并开启子进程的方式二
import time
import random
from multiprocessing import Process


class Mian(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print('%s name ' % self.name)
        time.sleep(random.randrange(1, 5))
        print('%s name ' % self.name)

if __name__ == '__main__':
    p1 = Mian('cc')
    p2 = Mian('zq')
    p1.start()
    p2.start()

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，可用来开启一个子进程 强调： 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

参数介绍：

group参数未使用，值始终为None target表示调用对象，即子进程要执行的任务 args表示调用对象的位置参数元组，args=('cc',) kwargs表示调用对象的字典,kwargs={'name':'cc','age':18} name为子进程的名称


方法介绍：

start()：启动进程，并调用该子进程中的p.run()

join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间。

run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法


terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁 is_alive():如果进程仍然运行，返回True


属性介绍：

daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置 name:进程的名称 pid：进程的pid

#!/urs/bin/evn python
# -*- coding:utf-8 -*-

import time
import random
from multiprocessing import Process


def task(name):
    print('%s is name' % name)
    time.sleep(random.randrange(1, 5))
    print('%s is name' % name)


if __name__ == '__main__':
    p1 = Process(target=task, args=('cc',))
    p1.start()
    p1.terminate() # 关闭进程，不会立即关闭is_alive立刻查看可能还是存活
    print(p1.is_alive())
    print('zhu')
    print(p1.is_alive())

#!/urs/bin/evn python
# -*- coding:utf-8 -*-

import os
import time
import random
from multiprocessing import Process


def task():
    print('%s is main' % os.getpid())
    time.sleep(random.randrange(1, 3))
    print('%s is main' % os.getpid())

if __name__ == '__main__':
    p1 = Process(target=task)
    p2 = Process(target=task)
    p3 = Process(target=task)
    p1.start()
    p1.join()
    p2.start()
    p2.join()
    p3.start()
    p3.join()
    print('zhu')

'''
terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
is_alive():如果进程仍然运行，返回True.


上述启动进程与join进程可以简写为

p_l=[p1,p2,p3]

for p in p_l:
    p.start()

for p in p_l:
    p.join()



'''

import time
import random
from multiprocessing import Process


def task(name):
    print('%s is name' % name)
    time.sleep(random.randrange(1, 5))
    print('%s is name end' % name)

if __name__ == '__main__':
    p1 = Process(target=task, args=('cc',), name='子进程1')  # 可以用关键参数来指定进程名
    p1.start()
    p1.join()

    print(p1.name, p1.pid,)

守护进程：


#!/urs/bin/evn python
# -*- coding:utf-8 -*-

import os
import time
import random
from multiprocessing import Process


def task(name):
    print('%s is name' % name)
    time.sleep(random.randrange(1, 3))
    print('%s is name' % name)


if __name__ == '__main__':
    p = Process(target=task, args=('cc',))
    p.daemon = True     # 一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
    p.start()
    p.join()
    print('zhu')

'''
其一：守护进程会在主进程代码执行结束后就终止

其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
'''

import os
from time import sleep
from multiprocessing import Process


# 子进程要执行的代码
def run_proc(name, age, **kwargs):
    for i in range(10):
        print('子进程运行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
        print(kwargs)
        sleep(0.5)

if __name__ == '__main__':
    print('父进程 %d.' % os.getpid())
    p = Process(target=run_proc, args=('test', 18), kwargs={"m": 20}) 
    print('子进程将要执行')
    p.start()
    sleep(1)
    p.terminate()
    p.join()
    print('子进程已结束')

#!/urs/bin/evn python
# -*- coding=utf-8 -*-
import time
import os
from multiprocessing import Process


# 两个子进程将会调用的两个方法
def worker_1(interval):
    print("worker_1,父进程(%s),当前进程(%s)" % (os.getppid(), os.getpid()))
    t_start = time.time()
    time.sleep(interval) # 程序将会被挂起interval秒
    t_end = time.time()
    print("worker_1,执行时间为'%0.2f'秒" % (t_end - t_start))


def worker_2(interval):
    print("worker_2,父进程(%s),当前进程(%s)" % (os.getppid(), os.getpid()))
    t_start = time.time()
    time.sleep(interval)
    t_end = time.time()
    print("worker_2,执行时间为'%0.2f'秒" % (t_end - t_start))
if __name__ == '__main__':

    # 输出当前程序的ID
    print("进程ID：%s" % os.getpid())

    # 创建两个进程对象，target指向这个进程对象要执行的对象名称，
    # args后面的元组中，是要传递给worker_1方法的参数，
    # 因为worker_1方法就一个interval参数，这里传递一个整数2给它，
    # 如果不指定name参数，默认的进程对象名称为Process-N，N为一个递增的整数
    p1 = Process(target=worker_1, args=(2,))
    p2 = Process(target=worker_2, name="dongGe",args=(1,))

    # 使用"进程对象名称.start()"来创建并执行一个子进程，
    # 这两个进程对象在start后，就会分别去执行worker_1和worker_2方法中的内容
    p1.start()
    p2.start()

    # 同时父进程仍然往下执行，如果p2进程还在执行，将会返回True
    print("p2.is_alive=%s" % p2.is_alive())

    # 输出p1和p2进程的别名和pid
    print("p1.name=%s" % p1.name)
    print("p1.pid=%s" % p1.pid)
    print("p2.name=%s" % p2.name)
    print("p2.pid=%s" % p2.pid)

    # join括号中不携带参数，表示父进程在这个位置要等待p1进程执行完成后，
    # 再继续执行下面的语句，一般用于进程间的数据同步，如果不写这一句，
    # 下面的is_alive判断将会是True，在shell（cmd）里面调用这个程序时
    # 可以完整的看到这个过程，大家可以尝试着将下面的这条语句改成p1.join(1)，
    # 因为p2需要2秒以上才可能执行完成，父进程等待1秒很可能不能让p1完全执行完成，
    # 所以下面的print会输出True，即p1仍然在执行
    p1.join()
    print("p1.is_alive=%s" % p1.is_alive())

进程池Pool

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来执行。

import os
import time, random
from  multiprocessing import Pool


def worker(msg):
    t_start = time.time()
    print("%s开始执行,进程号为%d" % (msg, os.getpid()))
    time.sleep((random.random()*2))
    t_stop = time.time()
    print(msg, "执行完毕，耗时%0.2f" % (t_stop - t_start))
if __name__ == '__main__':

    pl = Pool(3)  # 定义一个进程池，最大进程数为3
    for i in range(10):
        pl.apply_async(worker, (i,))
        # apply_async(要调用的目标,(传递给目标的参数元祖,))
        # 每次循环将会用空闲出来的子进程去调用目标
    print("----start----")
    pl.close()  # 关闭进程池，关闭后pl不再接收新的请求
    pl.join()  # 等待pl中所有子进程执行完成，必须放在close语句之后
    print("-----end-----")

multiprocessing.Pool常用函数解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；

• apply(func[, args[, kwds]])：使用阻塞方式调用func

• close()：关闭Pool，使其不再接受新的任务；

• terminate()：不管任务是否完成，立即终止；

• join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

!/urs/bin/evn python
# -*- coding:utf-8 -*-
import os
import time, random
from multiprocessing import Pool


def worker(msg):
    t_start = time.time()
    print("%s开始执行,进程号为%d" % (msg, os.getpid()))
    time.sleep((random.random()*2))
    t_stop = time.time()
    print(msg, "执行完毕，耗时%0.2f" % (t_stop - t_start))
if __name__ == '__main__':

    pl = Pool(3)  # 定义一个进程池，最大进程数为3
    for i in range(10):
        pl.apply(worker, (i,))
        # apply(要调用的目标,(传递给目标的参数元祖,))
        # 每次循环将会用空闲出来的子进程去调用目标
    print("----start----")
    pl.close()  # 关闭进程池，关闭后pl不再接收新的请求
    pl.join()  # 等待pl中所有子进程执行完成，必须放在close语句之后
    print("-----end-----")


全局变量或变量不共享数据


#!/urs/bin/evn python
# -*- coding:utf-8 -*-

import os
import time
from multiprocessing import Process


nums = [11, 22]


def work1():
    print('in process1 pid =  %d, nums=%s'% (os.getpid(), nums))
    for i in range(3):
        nums.append(i)
        time.sleep(3)
        print('in process1 pid =  %d, nums=%s'% (os.getpid(), nums))


def work2():
    print('in process2 pid =  %d, nums=%s'% (os.getpid(), nums))


if __name__ == '__main__':
    p1 = Process(target=work1)
    p2 = Process(target=work2)
    p1.start()
    p2.start()
    p1.join()
    p2.join()

互斥锁

　　进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争，竞争带来的结果就是错乱。互斥锁的工作原理就是多个人都要去争抢同一个资源，就是把并发改成行，降低了效率，但保证了数据安全不错乱

#!/urs/bin/evn python
# -*- coding:utf-8 -*-

import os
import time
from multiprocessing import Process


def work():
    print('%s is running' % os.getpid())
    time.sleep(3)
    print('%s is running' % os.getpid())


if __name__ == '__main__':
    for i in range(5):
        p = Process(target=work)
        p.start()

加锁：


!/urs/bin/evn python
# -*- coding:utf-8 -*-

import os
import time
from multiprocessing import Process, Lock


def work(lock):
    lock.acquire()  # 加锁
    print('%s is running' % os.getpid())
    time.sleep(2)
    print('%s is running' % os.getpid())
    lock.release()  # 释放锁


if __name__ == '__main__':
    lock = Lock()
    for i in range(3):
        p = Process(target=work, args=(lock,))
        p.start()

小列：

import json
import time
from multiprocessing import Process

# 文件db.txt的内容为：{"count":1}
# 注意一定要用双引号，不然json无法识别


def search(name):
    dic=json.load(open('db.txt'))
    print(dic)
    time.sleep(1)
    print('33[43m%s 查到剩余票数%s33[0m' %(name,dic['count']))
    

def get(name):
    dic = json.load(open('db.txt'))
    time.sleep(3)  # 模拟读数据的网络延迟
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(1)  # 模拟写数据的网络延迟
        json.dump(dic, open('db.txt', 'w'))
        print('33[46m%s 购票成功33[0m' % name)


def task(name):
    search(name)
    get(name)

if __name__ == '__main__':
    for i in range(10): # 模拟并发10个客户端抢票
        name = '<路人%s>' % i
        p = Process(target=task, args=(name,))
        p.start()

import time
import json
from multiprocessing import Process, Lock


def search(name):
    dic=json.load(open('db.txt'))

    time.sleep(1)
    print('33[43m%s 查到剩余票数%s33[0m' % (name, dic['count']))


def get(name):
    dic = json.load(open('db.txt'))
    time.sleep(1) # 模拟读数据的网络延迟
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(1)  # 模拟写数据的网络延迟
        json.dump(dic, open('db.txt', 'w'))
        print('33[46m%s 购票成功33[0m' % name)


def task(name, lock):
    search(name)
    with lock:   # 相当于lock.acquire(),执行完自代码块自动执行lock.release()
        get(name)

if __name__ == '__main__':
    lock = Lock()
    for i in range(10):  # 模拟并发10个客户端抢票
        name = '<路人%s>' % i
        p = Process(target=task, args=(name, lock))
        p.start()

加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行地修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。

虽然可以用文件共享数据实现进程间通信，但问题是：

1、效率低（共享数据基于文件，而文件是硬盘上的数据）

2、需要自己加锁处理

因此我们最好找寻一种解决方案能够兼顾：

1、效率高（多个进程共享一块内存的数据）

2、帮我们处理好锁问题。

这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。

队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信的最佳选择。

我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

创建队列的类（底层就是以管道和锁定的方式实现）：

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

参数介绍：

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

主要方法介绍：

初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最大可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）；

• Queue.qsize()：返回当前队列包含的消息数量；

• Queue.empty()：如果队列为空，返回True，反之False ；

• Queue.full()：如果队列满了，返回True,反之False；

• Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True；

• Queue.put():方法用以插入数据到队列中。

1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出"Queue.Empty"异常；

2）如果block值为False，消息列队如果为空，则会立刻抛出"Queue.Empty"异常；

• Queue.get_nowait()：相当Queue.get(False)；

• Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为True；

1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息列队腾出空间为止，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出"Queue.Full"异常；

2）如果block值为False，消息列队如果没有空间可写入，则会立刻抛出"Queue.Full"异常；

• Queue.put_nowait(item)：相当Queue.put(item, False)；

#!/urs/bin/evn python
# -*- coding:utf-8 -*-

from multiprocessing import Queue

q = Queue(3)
q.put(1)
q.put('a')
q.put(['b'])
# print(q.full())  # 判断队列是否满
'''
#因为消息列队已满下面的try都会抛出异常，第一个try会等待2秒后再抛出异常，第二个Try会立刻抛出异常
try:
    q.put("cc", True, 2)
except:
    print("消息列队已满，现有消息数量:%s" % q.qsize())

try:
    q.put_nowait('cc')
except:
    print('消息列队已满，现有消息数量:%s' % q.qsize())
'''

# 推荐的方式，先判断消息列队是否已满，再写入
if not q.full():
    q.put_nowait('cc')
# 读取消息时，先判断消息列队是否为空，再读取
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

队列例子：

#!/urs/bin/evn python
# -*- coding:utf-8 -*-
import os
import time, random
from multiprocessing import Process, Queue


def write(q):
    '''
    写数据进程执行的代码
    :param q: 
    :return: 
    '''

    for i in ['a', 'b', 'c']:
        print('put %s to queue' % i)
        q.put(i)
        time.sleep(random.random())


def read(q):
    """
    读数据进程执行的代码
    :param q: 
    :return: 
    """
    while True:
        if not q.get(True):
            print('get %s from queue' % i)
            time.sleep(random.random())
        else:
            break


if __name__ == '__main__':
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=write, args=(q,))
    # 启动子进程pw，写入:
    pw.start()
    # 等待pw结束:
    pw.join()
    # 启动子进程pr，读取:
    pr.start()
    pr.join()
    # pr进程里是死循环，无法等待其结束，只能强行终止:
    print()

    print('所有数据都写入并且读完')

mport os
import time, random
from multiprocessing import Pool


def worker(msg):
    t_start = time.time()
    print("%s开始执行,进程号为%d" % (msg, os.getpid()))
    time.sleep((random.random()*2))
    t_stop = time.time()
    print(msg, "执行完毕，耗时%0.2f" % (t_stop - t_start))
if __name__ == '__main__':

    pl = Pool(3)  # 定义一个进程池，最大进程数为3
    for i in range(10):
        pl.apply_async(worker, (i,))
        
        # apply_async(要调用的目标,(传递给目标的参数元祖,))
        # 每次循环将会用空闲出来的子进程去调用目标
    print("----start----")
    pl.close()  # 关闭进程池，关闭后pl不再接收新的请求
    pl.join()  # 等待pl中所有子进程执行完成，必须放在close语句之后
    print("-----end-----")

进程池中的Queue

一 生产者消费者模型介绍

为什么要使用生产者消费者模型

生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者和消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。

import time,random,os
from multiprocessing import Process,Queue


def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('33[43m%s 吃 %s33[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('33[45m%s 生产了 %s33[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者
    p1=Process(target=producer,args=(q,'cc','包子'))

    #消费者
    c1=Process(target=consumer,args=(q,'zq'))

    #开始
    p1.start()
    c1.start()
    print('主')

此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。

import os
import time, random
from multiprocessing import Process, Queue


def consumer(q, name):
    while True:
        res = q.get()
        if res is None:
            break
        time.sleep(random.randint(1, 3))
        print('33[41m%s 吃 %s33[0m' %(name, res))


def producer(q, name, food):
    '''
    生产者
    :param q: 
    :param name: 
    :param food: 
    :return: 
    '''
    for i in range(3):
        time.sleep(random.randint(1, 3))
        res = '%s%s' % (food, i)
        q.put(res)
        print('33[44m%s 生产了 %s33[0m' % (name, res))


if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=(q, 'cc','baozi'))
    c1 = Process(target=consumer, args=(q, 'zq'))
    p1.start()
    c1.start()
    p1.join()
     
    q.put(None)
    print('zhu')

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

import time,random,os
from multiprocessing import Process,Queue

def consumer(q,name):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3))
        print('33[43m%s 吃 %s33[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('33[45m%s 生产了 %s33[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们
    p1=Process(target=producer,args=(q,'cc','包子'))
    p2=Process(target=producer,args=(q,'zq','包子2'))
    p3=Process(target=producer,args=(q,'xqx','包子3'))

    #消费者们:
    c1=Process(target=consumer,args=(q,'mm1'))
    c2=Process(target=consumer,args=(q,'mm2'))

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    q.put(None)
    q.put(None)
    q.put(None)
    print('主')

JoinableQueue([maxsize])

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

参数介绍

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

方法介绍

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

import time, random, os
from multiprocessing import Process,JoinableQueue


def consumer(q, name):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('33[43m%s 吃 %s33[0m' %(name,res))
        q.task_done() # 发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了


def producer(q, name, food):
    for i in range(3):
        time.sleep(random.randint(1, 3))
        res = '%s%s' %(food,i)
        q.put(res)
        print('33[45m%s 生产了 %s33[0m' %(name,res))
    q.join()  # 等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束

if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer,args=(q, 'cc','baozi'))
    c1 = Process(target=consumer,args=(q, 'zq'))
    c1.daemon = True
    p1.start()
    c1.start()
    p1.join()
    print('zhu')