Python多进程Queue队列详解：参数、阻塞与非阻塞处理及生产者消费者模型

from multiprocessing import Queue
q = Queue() # 生成一个队列对象
# put方法是往队列里面放值
q.put('Cecilia陈')
q.put('xuchen')
q.put('喜陈')

# get方法是从队列里面取值
print(q.get())
print(q.get())
print(q.get())

q.put(5)
q.put(6)
print(q.get())

from multiprocessing import Queue
q = Queue(3)
q.put('Cecilia陈')
q.put('xuchen')
q.put('喜陈')
print(q.full()) # 判断队列是否满了 返回的是True/False
q.put(2) # 当队列已经满的时候，再放值,程序会阻塞，但不会结束

from multiprocessing import Queue
q = Queue(3)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=5) # 此时程序将对等待5秒以后报错了

from multiprocessing import Queue
q = Queue()
q.put('Cecilia陈')
print(q.get())
q.get(block=True,timeout=2) # 此时程序会等待2秒后，报错了，队列里面没有值了

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陈')
q.put('喜陈')
print(q.full())
q.put('xichen',block=False) # 队列已经满了，我不等待了，直接报错

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陈')
q.put('喜陈')
print(q.get())
print(q.get())
print(q.get(block=False)) # 队列已经没有值了，我不等待了，直接报错

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陈')
q.put('喜陈')
print(q.full())

q.put_nowait('xichen') # 程序不等待，不阻塞，直接报错

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陈')
q.put('喜陈')
print(q.get())
print(q.get())
print(q.full())
q.get_nowait()# 再取值的时候，程序不等待，不阻塞，程序直接报错

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

from multiprocessing import Queue
q=Queue(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 f(q,name,age):
  q.put(name,age) #调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。
if __name__ == '__main__':
  q = Queue() #创建一个Queue对象
  p = Process(target=f, args=(q,'Cecilia陈',18)) #创建一个进程
  p.start()
  print(q.get())
  p.join()

from multiprocessing import Queue,Process
# 生产者
def producer(q,name,food):
  for i in range(3):
    print(f'{name}生产了{food}{i}')
    res = f'{food}{i}'
    q.put(res)
# 消费者
def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    print(f'{name}吃了{res}')
if __name__ == '__main__':
  q = Queue() # 为的是让生产者和消费者使用同一个队列，使用同一个队列进行通讯
  p1 = Process(target=producer,args=(q,'Cecilia陈','巧克力'))
  c1 = Process(target=consumer,args=(q,'Tom'))
  p1.start()
  c1.start()

from multiprocessing import Queue,Process
def producer(q,name,food):
  for i in range(3):
    print(f'{name}生产了{food}{i}')
    res = f'{food}{i}'
    q.put(res)
  q.put(None) # 当生产者结束生产的的时候，我们再队列的最后再做一个表示，告诉消费者，生产者已经不生产了，让消费者不要再去队列里拿东西了
def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    if res == None:break # 判断队列拿出的是不是生产者放的结束生产的标识，如果是则不取，直接退出，结束程序
    print(f'{name}吃了{res}')
if __name__ == '__main__':
  q = Queue() # 为的是让生产者和消费者使用同一个队列，使用同一个队列进行通讯
  p1 = Process(target=producer,args=(q,'Cecilia陈','巧克力'))
  c1 = Process(target=consumer,args=(q,'Tom'))
  p1.start()
  c1.start()

from multiprocessing import Queue,Process
import time,random

def producer(q,name,food):
  for i in range(3):
    print(f'{name}生产了{food}{i}')
    time.sleep((random.randint(1,3)))
    res = f'{food}{i}'
    q.put(res)
  # q.put(None) # 当生产者结束生产的的时候，我们再队列的最后再做一个表示，告诉消费者，生产者已经不生产了，让消费者不要再去队列里拿东西了



def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    if res == None:break # 判断队列拿出的是不是生产者放的结束生产的标识，如果是则不取，直接退出，结束程序
    time.sleep((random.randint(1, 3)))
    print(f'{name}吃了{res}')

if __name__ == '__main__':
  q = Queue() # 为的是让生产者和消费者使用同一个队列，使用同一个队列进行通讯
  # 多个生产者进程
  p1 = Process(target=producer,args=(q,'Cecilia陈','巧克力'))
  p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
  p3 = Process(target=producer,args=(q,'喜陈','可乐'))
  # 多个消费者进程
  c1 = Process(target=consumer,args=(q,'Tom'))
  c2 = Process(target=consumer,args=(q,'jack'))


  # 告诉操作系统启动生产者进程
  p1.start()
  p2.start()
  p3.start()

  # 告诉操作系统启动消费者进程
  c1.start()
  c2.start()

  p1.join()
  p2.join()
  p3.join()

  q.put(None) # 几个消费者put几次
  q.put(None)

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

def producer(q,name,food):
  for i in range(3):
    print(f'{name}生产了{food}{i}')
    # time.sleep((random.randint(1,3)))
    res = f'{food}{i}'
    q.put(res)
  # q.put(None) # 当生产者结束生产的的时候，我们再队列的最后再做一个表示，告诉消费者，生产者已经不生产了，让消费者不要再去队列里拿东西了
  q.join()


def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    # if res == None:break # 判断队列拿出的是不是生产者放的结束生产的标识，如果是则不取，直接退出，结束程序
    # time.sleep((random.randint(1, 3)))
    print(f'{name}吃了{res}')
    q.task_done()#向q.join()发送一次信号,证明一个数据已经被取走了


if __name__ == '__main__':
  q = JoinableQueue() # 为的是让生产者和消费者使用同一个队列，使用同一个队列进行通讯
  # 多个生产者进程
  p1 = Process(target=producer,args=(q,'Cecilia陈','巧克力'))
  p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
  p3 = Process(target=producer,args=(q,'喜陈','可乐'))
  # 多个消费者进程
  c1 = Process(target=consumer,args=(q,'Tom'))
  c2 = Process(target=consumer,args=(q,'jack'))


  # 告诉操作系统启动生产者进程
  p1.start()
  p2.start()
  p3.start()

  # 把生产者设为守护进程
  c1.daemon = True
  c2.daemon = True
  # 告诉操作系统启动消费者进程
  c1.start()
  c2.start()

  p1.join()
  p2.join()
  p3.join() # 等待生产者生产完毕

  print('主进程')

  ### 分析
  # 生产者生产完毕--这是主进程最后一行代码结束--q.join()消费者已经取干净了,没有存在的意义了
  # 这是主进程最后一行代码结束,消费者已经取干净了,没有存在的意义了.守护进程的概念.

from multiprocessing import Process,Queue,JoinableQueue
q = JoinableQueue()
q.put('zhao') # 放队列里一个任务
q.put('qian')
print(q.get())
q.task_done() # 完成了一次任务
print(q.get())
q.task_done() # 完成了一次任务
q.join() #计数器不为0的时候 阻塞等待计数器为0后通过

# 想象成一个计数器 :put +1  task_done -1