# Queue([maxsize]) 
创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。




底层队列使用管道和锁定实现。代码实现：

代码实现

‘‘‘
multiprocessing模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递实现的,但是队列接口
‘‘‘

from multiprocessing import Queue
q = Queue(3)    # 只能往这个队列放3个值

# put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
# q.put(3)  # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。
            # 如果队列中的数据一直不被取走，程序就会永远停在这里。
try:
    q.put_nowait(3)    # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。
except:     # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。
    print(‘队列已经满了‘)

# 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。
print(q.full())    # 判断是否满了

print(q.get())
print(q.get())
print(q.get())
# print(q.get())   # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。
try:
    q.get_nowait(3)    # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。
except:    # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
    print(‘队列已经空了‘)

print(q.empty())   # 判断是否空了

from multiprocessing import Process, Queue

def consume(q):
    print(‘son-->‘, q.get())    # 取走队列一条数据
    q.put(‘abc‘)    # 给队列增加一条数据

if __name__ == ‘__main__‘:
    q = Queue()
    p = Process(target=consume, args=(q,))
    p.start()
    q.put({‘haha‘: 123})    # 给队列增加一条数据
    p.join()    # 等待子进程执行完毕
    print(‘Foo-->‘, q.get())    # 取走队列一条数据
# son--> {‘haha‘: 123}
# Foo--> abc 

import time
import random
from multiprocessing import Process, Queue

def consumer(q, name):
    # 处理数据
    while 1:
        food = q.get()
        if food is None: break  # 如果不结束的话程序就会一直不结束
        time.sleep(random.uniform(0.5, 1))
        print(‘%s吃了一个%s‘ % (name, food))

def producer(q, name, food):
    # 获取数据
    for i in range(10):
        time.sleep(random.uniform(0.3, 0.8))
        print(‘%s生产了%s%s‘ % (name, food, i + 1))
        q.put(food + str(i))

if __name__ == ‘__main__‘:
    q = Queue()
    Process(target=consumer, args=(q, ‘alex‘)).start()
    Process(target=consumer, args=(q, ‘wusir‘)).start()
    p1 = Process(target=producer, args=(q, ‘yang‘, ‘包子‘))
    p2 = Process(target=producer, args=(q, ‘sihao‘, ‘馒头‘))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    q.put(None)    # 有几个consumer就需要放几个None
    q.put(None)    # 结束信号 

import time
import random
from multiprocessing import Process, JoinableQueue
 
def consumer(q, name):
    # 处理数据
    while 1:
        food = q.get()
        time.sleep(random.uniform(0.5, 1))
        print(‘%s吃了一个%s‘ % (name, food))
        q.task_done()   # 通知队列已经有一个数据被处理了
        
def producer(q, name, food):
    # 获取数据
    for i in range(10):
        time.sleep(random.uniform(0.3, 0.8))
        print(‘%s生产了%s%s‘ % (name, food, i+1))
        q.put(food + str(i))
        
if __name__ == ‘__main__‘:
    q = JoinableQueue()
    c1 = Process(target=consumer, args=(q, ‘alex‘))
    c2 = Process(target=consumer, args=(q, ‘wusir‘))
    c1.daemon = True    # 设置守护进程
    c2.daemon = True    # 设置守护进程
    c1.start()
    c2.start()
    p1 = Process(target=producer, args=(q, ‘yang‘, ‘包子‘))
    p2 = Process(target=producer, args=(q, ‘sihao‘, ‘馒头‘))
    p1.start()
    p2.start()
    p1.join()   # 生产者要先把所有的数据都放到队列中
    p2.join()   # 生产者要先把所有的数据都放到队列中
    q.join()    # 阻塞直到放入队列中所有的数据都被处理掉(有多少个数据就接收到了多少task_done)

import time
from multiprocessing import Pipe, Process

def consumer(left, right):
    time.sleep(1)
    print(right.recv())

if __name__ == ‘__main__‘:
    left, right = Pipe()
    Process(target=consumer, args=(left, right)).start()
    left.send(1234)

from multiprocessing import Process, Pipe

def consumer(left, right):
    left.close()    # 不写close将不会引发EOFError
    while 1:
        try:
            print(right.recv())
        except EOFError:
            break

if __name__ == ‘__main__‘:
    left, right = Pipe()
    Process(target=consumer, args=(left, right)).start()
    right.close()
    for i in range(10):
        left.send(‘包子%s‘ % i)
    left.close()
# pipe的端口管理不会随着某一个进程的关闭就关闭
# 操作系统来管理进程对这些端口的使用
# left,right
# left,right
# 操作系统管理4个端口  每关闭一个端口计数-1，直到所有的端口都关闭了
# 剩余1个端口的时候 recv就会报错

from multiprocessing import Manager, Process, Lock

def work(d, lock):
    with lock:  # 不加锁而操作共享的数据，肯定会出现数据错乱
        d[‘count‘] -= 1
        
if __name__ == ‘__main__‘:
    lock = Lock()
    with Manager() as m:
        dic = m.dict({‘count‘: 100})
        p_l = []
        for i in range(100):
            p = Process(target=work, args=(dic, lock))
            p_l.append(p)
            p.start()
        for p in p_l:
            p.join()
        print(dic)

Pool([numprocess  [,initializer [, initargs]]]):创建进程池

# 参数介绍：
1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None
3 initargs：是要传给initializer的参数组

# 主要方法：
1 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
2 ‘‘‘需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，
必须从不同线程调用p.apply()函数或者使用p.apply_async()‘‘‘
3 p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
4 ‘‘‘此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，
将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。‘‘‘   
5 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
6 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

# 其他方法（了解）
1 方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法
2 obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。
如果远程操作中引发了异常，它将在调用此方法时再次被引发。
3 obj.ready():如果调用完成，返回True
4 obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
5 obj.wait([timeout]):等待结果变为可用。
6 obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数

# 进程池的 同步调用  apply

import os
import time
from multiprocessing import Pool

def task(num):
    time.sleep(1)
    print(‘%s : %s‘ % (num, os.getpid()))
    return num**2

if __name__ == ‘__main__‘:
    p = Pool()
    for i in range(20):
        res = p.apply(task, args=(i,))  # 提交任务的方法，同步提交
        print(‘-->‘, res)

# 进程池的 异步调用  apply_async

import os
import time
from multiprocessing import Pool
# 没有取返回值
def task(num):
    time.sleep(1)
    print(‘%s : %s‘ % (num, os.getpid()))
    return num**2

if __name__ == ‘__main__‘:
    p = Pool()
    for i in range(20):
        p.apply_async(task, args=(i,))  # 提交任务的方法，异步提交
    p.close()
    p.join()

# 通过队列取返回值
def task(num):
    time.sleep(1)
    print(‘%s : %s‘ % (num, os.getpid()))
    return num**2

if __name__ == ‘__main__‘:
    p = Pool()
    res_lst = []
    for i in range(20):
        res = p.apply_async(task, args=(i,))    # 提交任务的方法，异步提交
        res_lst.append(res)
    for res in res_lst:
        print(res.get())

import os
import time
from multiprocessing import Pool

def task(num):
    time.sleep(1)
    print(‘%s : %s‘ % (num, os.getpid()))
    return num**2

if __name__ == ‘__main__‘:
    p = Pool()
    p.map(task, range(20))