Python队列安全案例如何保障队列安全

wen python案例 2026-06-28 250

本文目录导读：

线程安全队列 - queue.Queue
进程安全队列 - multiprocessing.Queue
保障队列安全的核心机制
高级安全队列模式
性能与安全的平衡
第三方增强队列

在Python中,队列安全（多线程/多进程安全）主要通过标准库提供的线程安全队列和进程安全队列来实现，以下是保障队列安全的核心案例和原理：

线程安全队列 - queue.Queue

基础案例

from queue import Queue
import threading
import time
# 创建线程安全队列
safe_queue = Queue(maxsize=10)
# 生产者
def producer():
    for i in range(5):
        item = f"数据-{i}"
        safe_queue.put(item)  # 线程安全操作
        print(f"生产: {item}")
        time.sleep(0.5)
# 消费者
def consumer():
    while True:
        item = safe_queue.get()  # 线程安全操作
        print(f"消费: {item}")
        safe_queue.task_done()
        time.sleep(1)
# 启动线程
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()

进程安全队列 - multiprocessing.Queue

跨进程案例

from multiprocessing import Process, Queue
def worker(q):
    while True:
        task = q.get()
        if task is None:  # 终止信号
            break
        print(f"进程{os.getpid()} 处理 {task}")
if __name__ == "__main__":
    q = Queue()
    # 启动多个工作进程
    workers = []
    for _ in range(3):
        p = Process(target=worker, args=(q,))
        p.start()
        workers.append(p)
    # 添加任务
    for i in range(10):
        q.put(f"任务-{i}")
    # 发送终止信号（每个进程一个）
    for _ in workers:
        q.put(None)
    for p in workers:
        p.join()

保障队列安全的核心机制

① 内置锁机制

# Queue 内部实现（简化示意）
class SafeQueue:
    def __init__(self):
        self._mutex = threading.Lock()  # 互斥锁
        self._not_empty = threading.Condition(self._mutex)
        self._not_full = threading.Condition(self._mutex)
    def put(self, item, block=True, timeout=None):
        with self._not_full:  # 自动加锁
            while self.full():
                self._not_full.wait()  # 队列满时等待
            self._put(item)
            self._not_empty.notify()  # 通知消费者
    def get(self, block=True, timeout=None):
        with self._not_empty:  # 自动加锁
            while self.empty():
                self._not_empty.wait()  # 队列空时等待
            item = self._get()
            self._not_full.notify()  # 通知生产者
            return item

② 防止死锁设计

# 使用超时机制避免死锁
try:
    item = safe_queue.get(timeout=5)  # 5秒超时
except queue.Empty:
    print("队列为空，超时退出")
# 非阻塞操作
if not safe_queue.empty():
    item = safe_queue.get_nowait()  # 不等待

高级安全队列模式

批量处理+异常处理

from queue import Queue, Empty
import threading
class SafeTaskQueue:
    def __init__(self, maxsize=100):
        self.queue = Queue(maxsize=maxsize)
        self.running = threading.Event()
        self.running.set()
    def add_task(self, task, priority=0):
        """安全添加任务"""
        self.queue.put((priority, task), timeout=5)
    def get_batch(self, batch_size=10):
        """批量获取任务"""
        tasks = []
        for _ in range(batch_size):
            try:
                _, task = self.queue.get_nowait()
                tasks.append(task)
            except Empty:
                break
        return tasks
    def stop(self):
        """安全停止"""
        self.running.clear()
        # 清空队列防止阻塞
        while not self.queue.empty():
            try:
                self.queue.get_nowait()
            except Empty:
                break
# 使用示例
task_queue = SafeTaskQueue()
def worker():
    while task_queue.running.is_set():
        batch = task_queue.get_batch()
        if batch:
            for task in batch:
                # 处理任务
                pass
        else:
            threading.Event().wait(0.1)
# 启动工作线程
threads = [threading.Thread(target=worker) for _ in range(4)]
for t in threads:
    t.daemon = True  # 守护线程
    t.start()

性能与安全的平衡

使用队列最佳实践

import queue
import time
# 1. 合理设置队列大小
q = queue.Queue(maxsize=1000)  # 防止内存溢出
# 2. 使用 join() 确保所有任务完成
def worker():
    while True:
        item = q.get()
        if item is None:  # 哨兵值
            break
        # 处理任务
        q.task_done()
# 等待所有任务完成
q.join()
# 3. 监控队列状态
q_size = q.qsize()  # 当前队列大小
is_full = q.full()
is_empty = q.empty()
# 4. 优雅关闭
def shutdown_workers(workers, q):
    """安全关闭所有工作线程"""
    for _ in workers:
        q.put(None)  # 发送终止信号
    for w in workers:
        w.join(timeout=5)  # 超时等待

第三方增强队列

Redis队列（适合分布式场景）

import redis
from rq import Queue as RQueue
from rq.job import Job
# Redis 支持的队列
r = redis.Redis()
q = RQueue(connection=r)
# 安全特性
job = q.enqueue(func, args=(1, 2))
result = job.result  # 获取结果
status = job.get_status()  # 检查状态