Python脚本如何避免同步任务扎堆执行

wen python案例 2026-06-26 225

本文目录导读：

使用 asyncio.Semaphore（异步场景）
使用 concurrent.futures.ThreadPoolExecutor（多线程场景）
使用队列 + 消费者模式（生产消费模型）
使用第三方库 ratelimit 或 throttle
分布式场景：使用 Redis 实现分布式锁或队列
最佳实践建议

在Python中避免同步任务扎堆执行，核心思路是控制并发数和防止任务堆积,以下是几种常见场景的解决方案：

使用 `asyncio.Semaphore`（异步场景）

当使用 asyncio 时,信号量可以限制同时执行的任务数：

import asyncio
import aiohttp
async def worker(sem, url):
    async with sem:  # 获取信号量，如果已满则等待
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                data = await resp.text()
                print(f"处理 {url}: 长度 {len(data)}")
        await asyncio.sleep(0.1)  # 模拟其他耗时操作
async def main():
    sem = asyncio.Semaphore(5)  # 最多5个任务同时执行
    urls = [f"https://httpbin.org/get?id={i}" for i in range(100)]
    # 创建所有任务但会通过信号量控制并发
    tasks = [worker(sem, url) for url in urls]
    await asyncio.gather(*tasks)
asyncio.run(main())

使用 `concurrent.futures.ThreadPoolExecutor`（多线程场景）

限制线程池大小来避免扎堆：

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random
def task(item):
    """模拟一个耗时的同步任务"""
    delay = random.uniform(0.1, 0.5)
    time.sleep(delay)
    return f"任务 {item} 完成, 耗时 {delay:.2f}s"
# 控制并发数：最多3个线程同时执行
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务，不会立即阻塞
    futures = {executor.submit(task, i): i for i in range(20)}
    # 按完成顺序获取结果
    for future in as_completed(futures):
        item = futures[future]
        try:
            result = future.result()
            print(result)
        except Exception as e:
            print(f"任务 {item} 失败: {e}")

使用队列 + 消费者模式（生产消费模型）

适用于任务生成速度可能超过处理速度的场景：

from queue import Queue
from threading import Thread
import time
import random
class TaskConsumer:
    def __init__(self, num_workers=3, max_queue_size=20):
        self.queue = Queue(maxsize=max_queue_size)  # 限制队列大小
        self.num_workers = num_workers
        self.workers = []
    def start(self):
        for i in range(self.num_workers):
            t = Thread(target=self._worker, args=(i,), daemon=True)
            t.start()
            self.workers.append(t)
    def _worker(self, worker_id):
        while True:
            try:
                task = self.queue.get(timeout=5)  # 5秒超时退出
                print(f"Worker {worker_id}: 开始处理 {task}")
                time.sleep(random.uniform(0.2, 0.8))  # 模拟处理时间
                print(f"Worker {worker_id}: 完成 {task}")
                self.queue.task_done()
            except:
                break  # 超时退出
    def add_task(self, task):
        """添加任务，如果队列已满会阻塞"""
        self.queue.put(task)  # 如果队列满，这里会等待
        print(f"添加任务 {task}, 当前队列大小: {self.queue.qsize()}")
# 使用示例
consumer = TaskConsumer(num_workers=3, max_queue_size=10)
consumer.start()
# 模拟快速生成任务
for i in range(100):
    consumer.add_task(f"任务_{i}")
    time.sleep(0.05)  # 控制生成速度
print("所有任务已提交")
consumer.queue.join()  # 等待所有任务完成

使用第三方库 `ratelimit` 或 `throttle`

控制执行频率：

from ratelimit import limits, sleep_and_retry
import time
@sleep_and_retry
@limits(calls=5, period=1)  # 每秒最多5次调用
def limited_function(item):
    print(f"执行: {item}")
    time.sleep(0.3)
# 即使快速循环，也会被限速
for i in range(50):
    limited_function(i)

分布式场景：使用 Redis 实现分布式锁或队列

import redis
import time
import random
r = redis.Redis(host='localhost', port=6379, db=0)
def process_with_redis_queue(task_id):
    """从Redis列表取任务，最多同时处理5个"""
    # 如果活跃任务超过限制，等待
    while r.llen(f"active_tasks:worker_{task_id}") >= 5:
        print("等待中...")
        time.sleep(0.5)
    # 模拟处理
    print(f"处理任务 {task_id}")
    r.lpush(f"active_tasks:worker_{task_id}", task_id)
    time.sleep(random.uniform(0.5, 2))
    r.lrem(f"active_tasks:worker_{task_id}", 0, task_id)
    print(f"完成任务 {task_id}")
# 多个worker可以独立运行
for i in range(20):
    process_with_redis_queue("worker_1")

最佳实践建议

明确瓶颈类型：
- CPU密集 → 使用 ProcessPoolExecutor
- IO密集 → 使用 ThreadPoolExecutor 或 asyncio
- 外部服务限制 → 使用信号量 + 限速

监控任务堆积：

import logging
from queue import Queue
class MonitoredQueue(Queue):
    def put(self, item, block=True, timeout=None):
        if self.qsize() > self.maxsize * 0.8:
            logging.warning(f"队列接近满: {self.qsize()}/{self.maxsize}")
        super().put(item, block, timeout)

动态调整并发数：

class AdaptiveExecutor:
    def __init__(self, min_workers=2, max_workers=10):
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.current_workers = min_workers
    def adjust_workers(self, queue_size, processing_time):
        if queue_size > 50 and processing_time < 0.1:
            self.current_workers = min(self.max_workers, self.current_workers + 1)
        elif queue_size < 10 and self.current_workers > self.min_workers:
            self.current_workers -= 1

选择哪种方案取决于你的具体场景：