如何动态调整并发量呢？

技术

yuzhiblue

2025年1月20日 20:10

动态调整并发量是指在程序运行过程中，根据当前的任务负载、系统资源或外部条件，实时增加或减少并发任务的数量。在单线程异步编程中（如使用 Python 的 asyncio），可以通过以下方法实现动态调整并发量：

1. 使用信号量动态调整并发量

asyncio.Semaphore 的初始值可以在运行时动态修改，从而实现并发量的动态调整。

（1）示例代码

import asyncio

async def download_file(url, semaphore):
    async with semaphore:  # 获取信号量
        print(f"开始下载: {url} (并发量: {semaphore._value})")
        await asyncio.sleep(2)  # 模拟下载耗时
        print(f"下载完成: {url}")

async def adjust_concurrency(semaphore, new_value):
    print(f"调整并发量: {semaphore._value} -> {new_value}")
    while semaphore._value < new_value:
        semaphore.release()  # 增加并发量
    while semaphore._value > new_value:
        await semaphore.acquire()  # 减少并发量

async def main(urls):
    semaphore = asyncio.Semaphore(3)  # 初始并发量为 3

    # 启动下载任务
    tasks = [asyncio.create_task(download_file(url, semaphore)) for url in urls]

    # 动态调整并发量
    await asyncio.sleep(5)  # 5 秒后增加并发量
    await adjust_concurrency(semaphore, 5)

    await asyncio.sleep(5)  # 再过 5 秒后减少并发量
    await adjust_concurrency(semaphore, 2)

    await asyncio.gather(*tasks)  # 等待所有任务完成

urls = [f"http://example.com/file{i}" for i in range(20)]
asyncio.run(main(urls))

（2）代码说明

semaphore.release()：增加信号量的值，从而增加并发量。
semaphore.acquire()：减少信号量的值，从而减少并发量。
adjust_concurrency 函数：根据目标并发量动态调整信号量的值。

2. 使用任务队列动态调整并发量

通过动态调整任务队列中的 worker 数量，可以实现并发量的动态调整。

（1）示例代码

import asyncio

async def download_file(url):
    print(f"开始下载: {url}")
    await asyncio.sleep(2)  # 模拟下载耗时
    print(f"下载完成: {url}")

async def worker(queue):
    while True:
        url = await queue.get()  # 从队列中获取任务
        await download_file(url)
        queue.task_done()  # 标记任务完成

async def adjust_workers(workers, queue, target_concurrency):
    current_concurrency = len(workers)
    if target_concurrency > current_concurrency:
        # 增加 worker
        for _ in range(target_concurrency - current_concurrency):
            worker_task = asyncio.create_task(worker(queue))
            workers.append(worker_task)
        print(f"增加 worker，当前并发量: {target_concurrency}")
    elif target_concurrency < current_concurrency:
        # 减少 worker
        for _ in range(current_concurrency - target_concurrency):
            workers.pop().cancel()  # 取消多余的 worker
        print(f"减少 worker，当前并发量: {target_concurrency}")

async def main(urls):
    queue = asyncio.Queue()
    workers = []

    # 初始并发量为 3
    await adjust_workers(workers, queue, 3)

    # 添加任务到队列
    for url in urls:
        queue.put_nowait(url)

    # 动态调整并发量
    await asyncio.sleep(5)  # 5 秒后增加并发量
    await adjust_workers(workers, queue, 5)

    await asyncio.sleep(5)  # 再过 5 秒后减少并发量
    await adjust_workers(workers, queue, 2)

    await queue.join()  # 等待所有任务完成

    # 取消所有 worker
    for w in workers:
        w.cancel()

urls = [f"http://example.com/file{i}" for i in range(20)]
asyncio.run(main(urls))

（2）代码说明

adjust_workers 函数：根据目标并发量动态调整 worker 数量。
workers 列表：存储当前运行的 worker 任务。
queue.join()：等待所有任务完成。
worker_task.cancel()：取消多余的 worker 任务。

3. 动态调整并发量的触发条件

动态调整并发量的触发条件可以根据以下因素： - 任务队列长度：如果队列中积压的任务过多，可以增加并发量。 - 系统资源利用率：如果 CPU 或内存使用率较低，可以增加并发量。 - 外部条件：如目标服务器的响应时间、网络带宽等。

示例：根据队列长度动态调整并发量

async def monitor_and_adjust(queue, workers):
    while True:
        queue_size = queue.qsize()
        if queue_size > 10 and len(workers) < 10:  # 队列过长，增加并发量
            await adjust_workers(workers, queue, len(workers) + 1)
        elif queue_size < 5 and len(workers) > 3:  # 队列过短，减少并发量
            await adjust_workers(workers, queue, len(workers) - 1)
        await asyncio.sleep(1)  # 每秒检查一次