Python進(jìn)程間通信 multiProcessing Queue隊(duì)列實(shí)現(xiàn)詳解

from multiprocessing import Queue
q = Queue() # 生成一個(gè)隊(duì)列對(duì)象
# put方法是往隊(duì)列里面放值
q.put('Cecilia陳')
q.put('xuchen')
q.put('喜陳')

# get方法是從隊(duì)列里面取值
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()) # 判斷隊(duì)列是否滿了 返回的是True/False
q.put(2) # 當(dāng)隊(duì)列已經(jīng)滿的時(shí)候，再放值,程序會(huì)阻塞，但不會(huì)結(jié)束

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) # 此時(shí)程序?qū)?duì)等待5秒以后報(bào)錯(cuò)了

from multiprocessing import Queue
q = Queue()
q.put('Cecilia陳')
print(q.get())
q.get(block=True,timeout=2) # 此時(shí)程序會(huì)等待2秒后，報(bào)錯(cuò)了，隊(duì)列里面沒有值了

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.full())
q.put('xichen',block=False) # 隊(duì)列已經(jīng)滿了，我不等待了，直接報(bào)錯(cuò)

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.get())
print(q.get())
print(q.get(block=False)) # 隊(duì)列已經(jīng)沒有值了，我不等待了，直接報(bào)錯(cuò)

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

q.put_nowait('xichen') # 程序不等待，不阻塞，直接報(bào)錯(cuò)

from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.get())
print(q.get())
print(q.full())
q.get_nowait()# 再取值的時(shí)候，程序不等待，不阻塞，程序直接報(bào)錯(cuò)

'''
multiprocessing模塊支持進(jìn)程間通信的兩種主要形式:管道和隊(duì)列
都是基于消息傳遞實(shí)現(xiàn)的,但是隊(duì)列接口
'''

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)  # 如果隊(duì)列已經(jīng)滿了，程序就會(huì)停在這里，等待數(shù)據(jù)被別人取走，再將數(shù)據(jù)放入隊(duì)列。
      # 如果隊(duì)列中的數(shù)據(jù)一直不被取走，程序就會(huì)永遠(yuǎn)停在這里。
try:
  q.put_nowait(3) # 可以使用put_nowait，如果隊(duì)列滿了不會(huì)阻塞，但是會(huì)因?yàn)殛?duì)列滿了而報(bào)錯(cuò)。
except: # 因此我們可以用一個(gè)try語(yǔ)句來處理這個(gè)錯(cuò)誤。這樣程序不會(huì)一直阻塞下去，但是會(huì)丟掉這個(gè)消息。
  print('隊(duì)列已經(jīng)滿了')

# 因此，我們?cè)俜湃霐?shù)據(jù)之前，可以先看一下隊(duì)列的狀態(tài)，如果已經(jīng)滿了，就不繼續(xù)put了。
print(q.full()) #滿了
print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一樣，如果隊(duì)列已經(jīng)空了，那么繼續(xù)取就會(huì)出現(xiàn)阻塞。
try:
  q.get_nowait(3) # 可以使用get_nowait，如果隊(duì)列滿了不會(huì)阻塞，但是會(huì)因?yàn)闆]取到值而報(bào)錯(cuò)。
except: # 因此我們可以用一個(gè)try語(yǔ)句來處理這個(gè)錯(cuò)誤。這樣程序不會(huì)一直阻塞下去。
  print('隊(duì)列已經(jīng)空了')

print(q.empty()) #空了

from multiprocessing import Process, Queue
def f(q,name,age):
  q.put(name,age) #調(diào)用主函數(shù)中p進(jìn)程傳遞過來的進(jìn)程參數(shù) put函數(shù)為向隊(duì)列中添加一條數(shù)據(jù)。
if __name__ == '__main__':
  q = Queue() #創(chuàng)建一個(gè)Queue對(duì)象
  p = Process(target=f, args=(q,'Cecilia陳',18)) #創(chuàng)建一個(gè)進(jìn)程
  p.start()
  print(q.get())
  p.join()

from multiprocessing import Queue,Process
# 生產(chǎn)者
def producer(q,name,food):
  for i in range(3):
    print(f'{name}生產(chǎn)了{(lán)food}{i}')
    res = f'{food}{i}'
    q.put(res)
# 消費(fèi)者
def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    print(f'{name}吃了{(lán)res}')
if __name__ == '__main__':
  q = Queue() # 為的是讓生產(chǎn)者和消費(fèi)者使用同一個(gè)隊(duì)列，使用同一個(gè)隊(duì)列進(jìn)行通訊
  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}生產(chǎn)了{(lán)food}{i}')
    res = f'{food}{i}'
    q.put(res)
  q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時(shí)候，我們?cè)訇?duì)列的最后再做一個(gè)表示，告訴消費(fèi)者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費(fèi)者不要再去隊(duì)列里拿東西了
def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    if res == None:break # 判斷隊(duì)列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識(shí)，如果是則不取，直接退出，結(jié)束程序
    print(f'{name}吃了{(lán)res}')
if __name__ == '__main__':
  q = Queue() # 為的是讓生產(chǎn)者和消費(fèi)者使用同一個(gè)隊(duì)列，使用同一個(gè)隊(duì)列進(jìn)行通訊
  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}生產(chǎn)了{(lán)food}{i}')
    time.sleep((random.randint(1,3)))
    res = f'{food}{i}'
    q.put(res)
  # q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時(shí)候，我們?cè)訇?duì)列的最后再做一個(gè)表示，告訴消費(fèi)者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費(fèi)者不要再去隊(duì)列里拿東西了



def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    if res == None:break # 判斷隊(duì)列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識(shí)，如果是則不取，直接退出，結(jié)束程序
    time.sleep((random.randint(1, 3)))
    print(f'{name}吃了{(lán)res}')

if __name__ == '__main__':
  q = Queue() # 為的是讓生產(chǎn)者和消費(fèi)者使用同一個(gè)隊(duì)列，使用同一個(gè)隊(duì)列進(jìn)行通訊
  # 多個(gè)生產(chǎn)者進(jìn)程
  p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
  p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
  p3 = Process(target=producer,args=(q,'喜陳','可樂'))
  # 多個(gè)消費(fèi)者進(jìn)程
  c1 = Process(target=consumer,args=(q,'Tom'))
  c2 = Process(target=consumer,args=(q,'jack'))


  # 告訴操作系統(tǒng)啟動(dòng)生產(chǎn)者進(jìn)程
  p1.start()
  p2.start()
  p3.start()

  # 告訴操作系統(tǒng)啟動(dòng)消費(fèi)者進(jìn)程
  c1.start()
  c2.start()

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

  q.put(None) # 幾個(gè)消費(fèi)者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}生產(chǎn)了{(lán)food}{i}')
    # time.sleep((random.randint(1,3)))
    res = f'{food}{i}'
    q.put(res)
  # q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時(shí)候，我們?cè)訇?duì)列的最后再做一個(gè)表示，告訴消費(fèi)者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費(fèi)者不要再去隊(duì)列里拿東西了
  q.join()


def consumer(q,name):
  while True:
    res = q.get(timeout=5)
    # if res == None:break # 判斷隊(duì)列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識(shí)，如果是則不取，直接退出，結(jié)束程序
    # time.sleep((random.randint(1, 3)))
    print(f'{name}吃了{(lán)res}')
    q.task_done()#向q.join()發(fā)送一次信號(hào),證明一個(gè)數(shù)據(jù)已經(jīng)被取走了


if __name__ == '__main__':
  q = JoinableQueue() # 為的是讓生產(chǎn)者和消費(fèi)者使用同一個(gè)隊(duì)列，使用同一個(gè)隊(duì)列進(jìn)行通訊
  # 多個(gè)生產(chǎn)者進(jìn)程
  p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
  p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
  p3 = Process(target=producer,args=(q,'喜陳','可樂'))
  # 多個(gè)消費(fèi)者進(jìn)程
  c1 = Process(target=consumer,args=(q,'Tom'))
  c2 = Process(target=consumer,args=(q,'jack'))


  # 告訴操作系統(tǒng)啟動(dòng)生產(chǎn)者進(jìn)程
  p1.start()
  p2.start()
  p3.start()

  # 把生產(chǎn)者設(shè)為守護(hù)進(jìn)程
  c1.daemon = True
  c2.daemon = True
  # 告訴操作系統(tǒng)啟動(dòng)消費(fèi)者進(jìn)程
  c1.start()
  c2.start()

  p1.join()
  p2.join()
  p3.join() # 等待生產(chǎn)者生產(chǎn)完畢

  print('主進(jìn)程')

  ### 分析
  # 生產(chǎn)者生產(chǎn)完畢--這是主進(jìn)程最后一行代碼結(jié)束--q.join()消費(fèi)者已經(jīng)取干凈了,沒有存在的意義了
  # 這是主進(jìn)程最后一行代碼結(jié)束,消費(fèi)者已經(jīng)取干凈了,沒有存在的意義了.守護(hù)進(jìn)程的概念.

from multiprocessing import Process,Queue,JoinableQueue
q = JoinableQueue()
q.put('zhao') # 放隊(duì)列里一個(gè)任務(wù)
q.put('qian')
print(q.get())
q.task_done() # 完成了一次任務(wù)
print(q.get())
q.task_done() # 完成了一次任務(wù)
q.join() #計(jì)數(shù)器不為0的時(shí)候 阻塞等待計(jì)數(shù)器為0后通過

# 想象成一個(gè)計(jì)數(shù)器 :put +1  task_done -1