Python脚本如何动态调整异步并发量

本文目录导读：

1. 使用信号量（Semaphore）控制并发
2. 使用队列 + 工作线程模式
3. 基于系统负载的自适应调整
4. 使用第三方库：aiolimiter
5. 关键要点

在Python中动态调整异步并发量,主要有以下几种方法：

使用信号量（Semaphore）控制并发

import asyncio
import random
class DynamicConcurrencyController:
    def __init__(self, initial_concurrency=5):
        self.semaphore = asyncio.Semaphore(initial_concurrency)
        self.current_concurrency = initial_concurrency
        self.max_concurrency = 20
    async def adjust_concurrency(self, new_limit):
        """动态调整并发量"""
        old_limit = self.current_concurrency
        self.current_concurrency = min(new_limit, self.max_concurrency)
        if new_limit > old_limit:
            # 增加并发：创建新的信号量
            self.semaphore = asyncio.Semaphore(self.current_concurrency)
        # 注意：减少并发需要等待当前任务完成
        print(f"并发数调整: {old_limit} -> {self.current_concurrency}")
    async def process_task(self, task_id):
        async with self.semaphore:
            print(f"处理任务 {task_id}，当前并发: {self.current_concurrency}")
            # 模拟异步任务
            await asyncio.sleep(random.uniform(0.5, 2))
            return f"任务 {task_id} 完成"
    async def run(self, tasks):
        """运行任务集合"""
        # 启动一个后台任务定期调整并发
        adjust_task = asyncio.create_task(self.auto_adjust())
        # 处理所有任务
        results = []
        for task_id in tasks:
            result = await self.process_task(task_id)
            results.append(result)
        adjust_task.cancel()
        return results
    async def auto_adjust(self):
        """自动调整并发量的示例"""
        while True:
            await asyncio.sleep(3)
            # 模拟根据系统负载调整
            new_limit = random.randint(3, 10)
            await self.adjust_concurrency(new_limit)
# 使用示例
async def main():
    controller = DynamicConcurrencyController(initial_concurrency=3)
    tasks = list(range(1, 12))
    results = await controller.run(tasks)
    print("所有任务完成:", results)
# asyncio.run(main())

使用队列 + 工作线程模式

import asyncio
from collections import deque
class DynamicAsyncPool:
    def __init__(self, max_workers=10, min_workers=1):
        self.max_workers = max_workers
        self.min_workers = min_workers
        self.current_workers = min_workers
        self.task_queue = asyncio.Queue()
        self.workers = []
        self.running = True
        # 启动初始工作线程
        for _ in range(min_workers):
            self._start_worker()
    def _start_worker(self):
        """启动一个工作线程"""
        worker = asyncio.create_task(self._worker_loop())
        self.workers.append(worker)
    async def _worker_loop(self):
        """工作线程主循环"""
        while self.running:
            try:
                task = await asyncio.wait_for(
                    self.task_queue.get(), 
                    timeout=1.0
                )
                try:
                    await task
                finally:
                    self.task_queue.task_done()
            except asyncio.TimeoutError:
                continue
    async def adjust_workers(self, target_count):
        """调整工作线程数量"""
        target_count = max(self.min_workers, 
                          min(self.max_workers, target_count))
        while len(self.workers) < target_count:
            self._start_worker()
            print(f"添加工作线程，当前: {len(self.workers)}")
        while len(self.workers) > target_count:
            worker = self.workers.pop()
            worker.cancel()
            print(f"移除工作线程，当前: {len(self.workers)}")
        self.current_workers = len(self.workers)
    async def submit_task(self, coro):
        """提交任务"""
        await self.task_queue.put(coro)
    async def wait_completion(self):
        """等待所有任务完成"""
        await self.task_queue.join()
# 使用示例
async def example_task(task_id, delay):
    print(f"开始任务 {task_id}")
    await asyncio.sleep(delay)
    print(f"完成任务 {task_id}")
    return task_id
async def main():
    pool = DynamicAsyncPool(max_workers=10, min_workers=2)
    # 提交多个任务
    for i in range(20):
        task = example_task(i, 1.0)
        await pool.submit_task(task)
    # 动态调整并发
    await asyncio.sleep(2)
    await pool.adjust_workers(5)  # 增加到5个worker
    await asyncio.sleep(2)
    await pool.adjust_workers(3)  # 减少到3个worker
    await pool.wait_completion()
    print("所有任务完成")
# asyncio.run(main())

基于系统负载的自适应调整

import asyncio
import psutil  # 需要安装: pip install psutil
import time
class AdaptiveConcurrencyManager:
    def __init__(self, 
                 min_concurrency=1, 
                 max_concurrency=50,
                 target_cpu_usage=70.0,
                 target_memory_usage=80.0):
        self.min_concurrency = min_concurrency
        self.max_concurrency = max_concurrency
        self.target_cpu_usage = target_cpu_usage
        self.target_memory_usage = target_memory_usage
        self.current_concurrency = min_concurrency
        self.semaphore = asyncio.Semaphore(min_concurrency)
        self.adjusting = False
    async def monitor_and_adjust(self):
        """监控系统资源并动态调整"""
        while True:
            if not self.adjusting:
                self.adjusting = True
                try:
                    # 获取系统资源使用情况
                    cpu_percent = psutil.cpu_percent(interval=1)
                    memory_percent = psutil.virtual_memory().percent
                    # 根据负载调整并发
                    new_concurrency = self._calculate_optimal_concurrency(
                        cpu_percent, memory_percent
                    )
                    if new_concurrency != self.current_concurrency:
                        await self._apply_concurrency_change(new_concurrency)
                    print(f"CPU: {cpu_percent}%, 内存: {memory_percent}%, "
                          f"并发: {self.current_concurrency}")
                finally:
                    self.adjusting = False
            await asyncio.sleep(5)  # 每5秒检查一次
    def _calculate_optimal_concurrency(self, cpu_percent, memory_percent):
        """计算最优并发数"""
        usage_ratio = max(cpu_percent / self.target_cpu_usage,
                         memory_percent / self.target_memory_usage)
        if usage_ratio > 1.0:
            # 负载过高，减少并发
            new_concurrency = int(self.current_concurrency / usage_ratio)
        else:
            # 负载可接受，考虑增加并发
            ratio = 1.0 / usage_ratio if usage_ratio > 0 else 1.0
            new_concurrency = int(self.current_concurrency * min(ratio, 1.5))
        return max(self.min_concurrency, 
                   min(self.max_concurrency, new_concurrency))
    async def _apply_concurrency_change(self, new_concurrency):
        """应用并发数变更"""
        old_concurrency = self.current_concurrency
        if new_concurrency > old_concurrency:
            # 增加并发
            self.semaphore = asyncio.Semaphore(new_concurrency)
            self.current_concurrency = new_concurrency
            print(f"增加并发: {old_concurrency} -> {new_concurrency}")
        elif new_concurrency < old_concurrency:
            # 减少并发需要等待
            print(f"准备减少并发: {old_concurrency} -> {new_concurrency}")
            self.current_concurrency = new_concurrency
            # 在实际应用中，这里需要更复杂的逻辑来等待活跃任务完成
    async def execute_with_limit(self, task_func, *args, **kwargs):
        """在并发限制下执行任务"""
        async with self.semaphore:
            return await task_func(*args, **kwargs)
# 使用示例
async def main():
    manager = AdaptiveConcurrencyManager(
        min_concurrency=1,
        max_concurrency=10,
        target_cpu_usage=70.0
    )
    # 启动监控任务
    monitor_task = asyncio.create_task(manager.monitor_and_adjust())
    async def cpu_intensive_task(task_id):
        """模拟CPU密集型任务"""
        print(f"开始任务 {task_id}")
        # 模拟计算
        for _ in range(10**6):
            _ = 2 ** 10
        await asyncio.sleep(0.5)
        print(f"完成任务 {task_id}")
        return task_id
    # 执行多个任务
    tasks = []
    for i in range(20):
        tasks.append(manager.execute_with_limit(cpu_intensive_task, i))
    results = await asyncio.gather(*tasks)
    monitor_task.cancel()
    print(f"所有任务完成: {results}")
# asyncio.run(main())

使用第三方库：aiolimiter

import asyncio
from aiolimiter import AsyncLimiter  # 需要安装: pip install aiolimiter
async def dynamic_limiter_example():
    """使用aiolimiter实现动态限流"""
    limiter = AsyncLimiter(5, 1)  # 初始每秒5个请求
    async def adjust_rate(new_rate):
        """动态调整速率限制"""
        limiter.max_rate = new_rate
        # 注意：aiolimiter不支持直接改变rate，这里需要重建
        nonlocal limiter
        limiter = AsyncLimiter(new_rate, 1)
        print(f"调整速率: {new_rate}")
    async def task(item):
        # 使用限流器控制并发
        async with limiter:
            print(f"处理项目: {item}")
            await asyncio.sleep(0.2)
            return item
    # 模拟动态调整
    tasks = [task(i) for i in range(10)]
    await asyncio.sleep(1)
    await adjust_rate(10)  # 增加到每秒10个
    results = await asyncio.gather(*tasks)
    print(f"结果: {results}")
# asyncio.run(dynamic_limiter_example())

关键要点

1. 安全调整：增加并发相对安全，但减少并发时需要等待活跃任务完成
2. 资源监控：结合实际系统资源（CPU、内存、网络）动态调整
3. 平滑过渡：避免频繁大幅调整，使用渐进式变化
4. 回调机制：可以在并发调整前后添加回调函数

选择哪种方法取决于你的具体需求：

• 简单的并发控制：使用Semaphore
• 需要队列管理：使用工作线程模式
• 系统自适应：使用资源监控方案
• 速率限制：使用aiolimiter等专业库