进程间通信Queue详解 - Python生产者消费者模式

from multiprocessing import Process, Queue

# 创建队列
q = Queue(maxsize=0)

# 参数说明：
# maxsize: 队列最大容量，0表示无限制

# 常用方法：
# q.put(item, block=True, timeout=None)  — 放入元素
# q.get(block=True, timeout=None)        — 取出元素
# q.qsize()                              — 队列当前大小
# q.empty()                              — 判断是否为空
# q.full()                               — 判断是否已满
# q.close()                              — 关闭队列

# JoinableQueue 额外方法：
# q.task_done()                          — 标记一个任务完成
# q.join()                               — 等待所有任务完成

from multiprocessing import Queue

q = Queue()

# 基本用法
q.put("消息1")
q.put("消息2")
q.put(100)               # 可以放入任意可序列化对象
q.put({"key": "value"})

# 阻塞模式（队列满时等待）
q.put("数据", block=True, timeout=5)  # 5秒超时

# 非阻塞模式
try:
    q.put("数据", block=False)
except:
    print("队列已满")

from multiprocessing import Queue

q = Queue()
q.put("第一条消息")
q.put("第二条消息")

# 基本用法（阻塞等待）
msg = q.get()
print(msg)  # 输出: 第一条消息

# 带超时的 get
try:
    msg = q.get(timeout=3)
    print(msg)
except:
    print("超时，未获取到数据")

# 非阻塞 get
try:
    msg = q.get(block=False)
    print(msg)
except:
    print("队列为空")

from multiprocessing import Process, Queue
import time
import random

def producer(queue, items):
    """生产者：生产数据并放入队列"""
    for item in items:
        print(f"生产: {item}")
        queue.put(item)
        time.sleep(random.uniform(0.1, 0.5))
    queue.put(None)  # 发送结束信号

def consumer(queue):
    """消费者：从队列取出数据并处理"""
    while True:
        item = queue.get()
        if item is None:  # 收到结束信号
            break
        print(f"消费: {item}")
        time.sleep(random.uniform(0.2, 0.8))
    print("消费者结束")

if __name__ == '__main__':
    q = Queue()
    
    p = Process(target=producer, args=(q, ['苹果', '香蕉', '橙子', '葡萄']))
    c = Process(target=consumer, args=(q,))
    
    c.start()
    p.start()
    
    p.join()
    c.join()
    
    print("生产者-消费者模式执行完毕")

from multiprocessing import Process, Queue
import time
import random

def producer(queue, name, count):
    for i in range(count):
        item = f"{name}-产品{i}"
        queue.put(item)
        print(f"[{name}] 生产了 {item}")
        time.sleep(random.uniform(0.1, 0.3))

def consumer(queue, name):
    while True:
        try:
            item = queue.get(timeout=2)
            print(f"[{name}] 消费了 {item}")
            time.sleep(random.uniform(0.1, 0.5))
        except:
            print(f"[{name}] 队列为空，退出")
            break

if __name__ == '__main__':
    q = Queue(maxsize=10)
    
    producers = [
        Process(target=producer, args=(q, f'工厂{i}', 5))
        for i in range(2)
    ]
    consumers = [
        Process(target=consumer, args=(q, f'消费者{i}'))
        for i in range(3)
    ]
    
    for c in consumers:
        c.start()
    for p in producers:
        p.start()
    
    for p in producers:
        p.join()
    for c in consumers:
        c.join()