def put(self, obj, block=True, timeout=None):插入數(shù)據(jù)到隊(duì)列中
Block值默認(rèn)為True，代表當(dāng)隊(duì)列已滿時(shí)，會(huì)阻塞。如果block為False，則隊(duì)列滿會(huì)報(bào)異常Queue.Full
timeout表示會(huì)阻塞到指定時(shí)間，直到有剩余的空間供插入，如果時(shí)間超時(shí)，則報(bào)異常Queue.Full
def get(self, block=True, timeout=None):從隊(duì)列中取出數(shù)據(jù)
Block值默認(rèn)為True，代表當(dāng)隊(duì)列為空時(shí)，會(huì)阻塞。如果block為False，則隊(duì)列空會(huì)報(bào)異常Queue.Empty
timeout表示會(huì)等待到指定時(shí)間，直到取出數(shù)據(jù)，如果時(shí)間超時(shí)，則報(bào)異常Queue.Empty
def empty(self): 判斷隊(duì)列是否為空，如果空返回True
def full(self): 判斷隊(duì)列是否已滿，如果滿返回True
def qsize(self): 返回隊(duì)列的大小

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é)束")
 
 

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ì)列中的所有元素
    # 開啟進(jìn)程
    prod.start()
    con.start()
    prod.join()  # 等待生產(chǎn)和消費(fèi)完成，主線程結(jié)束
    print("結(jié)束")