本文目录导读：
在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())
关键要点
- 安全调整：增加并发相对安全，但减少并发时需要等待活跃任务完成
- 资源监控：结合实际系统资源（CPU、内存、网络）动态调整
- 平滑过渡：避免频繁大幅调整，使用渐进式变化
- 回调机制：可以在并发调整前后添加回调函数
选择哪种方法取决于你的具体需求：
- 简单的并发控制：使用Semaphore
- 需要队列管理：使用工作线程模式
- 系统自适应：使用资源监控方案
- 速率限制：使用aiolimiter等专业库
