from multiprocessing import Process, Manager
q = Manager().Queue(2)
q.put(1)
q.put(2,block=False,timeout=2)
def func():
? ? print(q.get())
?
p = Process(target=func)
print("size",q.qsize())
print("full",q.full())
p.start()
p.join()
print("empty",q.empty())
print("get", q.get())
print("get", q.get(block=False,timeout=2))

from multiprocessing import Process,Manager,active_children
import random
import queue
import time
?
class Producer(Process):
?
? ? def __init__(self,queue):
? ? ? ? super().__init__()
? ? ? ? self.queue = queue
?
? ? def run(self):
? ? ? ? for i in range(6):
? ? ? ? ? ? r = random.randint(0, 99)
? ? ? ? ? ? time.sleep(1)
? ? ? ? ? ? self.queue.put(r)
? ? ? ? ? ? print("add data{}".format(r))
?
class Consumer(Process):
?
? ? def __init__(self,queue):
? ? ? ? super().__init__()
? ? ? ? self.queue = queue
?
? ? def run(self):
? ? ? ? while True:
? ? ? ? ? if not self.queue.empty():
? ? ? ? ? ? ? ? data = self.queue.get()
? ? ? ? ? ? ? ? print("minus data{}".format(data))
?
?
if __name__ == '__main__':
? ? q = Manager().Queue() # 創(chuàng)建隊(duì)列
? ? p = Producer(q)
? ? c = Consumer(q)
? ? p.start()
? ? c.start()
? ? print(active_children()) ?# 查看現(xiàn)有的進(jìn)程
? ? p.join()
? ? c.join()
? ? print("結(jié)束")
?
?
>>>輸出
[<ForkProcess(SyncManager-1, started)>, <Producer(Producer-2, started)>, <Consumer(Consumer-3, started)>]
add data83
minus data83
add data72
minus data72
add data8
minus data8
add data63
minus data63
add data75
minus data75
add data52
minus data52

from multiprocessing import Process,JoinableQueue
import os
import time
import random
?
?
def print_log(msg, log_type="prod"):
? ? if log_type == 'prod':
? ? ? ? print("\033[32;1m%s\033[0m" %msg)
? ? elif log_type == 'con':
? ? ? ? print("\033[31;1m%s\033[0m" %msg)
?
def producer(q):
? ? """
? ? 生產(chǎn)者
? ? :param q:?
? ? :return:?
? ? """
? ? for i in range(10):
? ? ? ? data = random.randint(1,200)
? ? ? ? time.sleep(2)
? ? ? ? q.put(data) ?# 放入隊(duì)列
? ? ? ? msg = "add data {}".format(data)
? ? ? ? print_log(msg)
? ? q.join() ?# 生產(chǎn)者調(diào)用此方法進(jìn)行阻塞，直到隊(duì)列中所有的項(xiàng)目均被處理。
? ? # 阻塞將持續(xù)到隊(duì)列中的每個(gè)項(xiàng)目均調(diào)用q.task_done（）方法為止
?
?
?
?
def consumer(q):
? ? """
? ? 消費(fèi)者
? ? :param q:?
? ? :return:?
? ? """
? ? while True:
? ? ? ? if not q.empty():
? ? ? ? ? ? time.sleep(5)
? ? ? ? ? ? data = q.get()
? ? ? ? ? ? msg = "minus data{}".format(data)
? ? ? ? ? ? print_log(msg,"con")
? ? ? ? ? ? q.task_done() ?# q.get()的返回項(xiàng)目已經(jīng)被處理
?
?
if __name__ == '__main__':
? ? q = JoinableQueue()
? ? prod = Process(target=producer, args=(q,))
? ? con = Process(target=consumer, args=(q,))
? ? con.daemon = True ?# 設(shè)置為守護(hù)進(jìn)程，但是不用擔(dān)心，producer內(nèi)調(diào)用q.join保證了consumer已經(jīng)處理完隊(duì)列中的所有元素
? ? # 開(kāi)啟進(jìn)程
? ? prod.start()
? ? con.start()
?
? ? prod.join() ?# 等待生產(chǎn)和消費(fèi)完成，主線程結(jié)束
? ? print("結(jié)束")