如何在运行中的AsyncIO主循环中动态添加任务以维持并发数？

要实现始终保持4个任务并行运行、完成一个就补充新任务的需求，可以通过信号量控制并发+动态任务集合管理来实现，以下是针对你场景的完整解决方案：

核心思路 #

1. 用asyncio.Semaphore限制同时运行的任务数（设为4），确保不超过40核的资源上限（每个任务分配10核）。
2. 维护一个待运行/运行中任务的集合，每次有任务完成后，立即从任务池（或随机生成）中补充新任务，直到所有任务处理完毕。

模拟场景代码（适配你的counter协程） #

import asyncio
import random

async def counter(n):
    print(f'counter with argument {n} has been launched')
    # 模拟计算任务
    for i in range(n):
        for j in range(n):
            for k in range(n):
                pass
    await asyncio.sleep(0)
    print(f'counter with argument {n} has FINISHED')
    return n

# 用信号量包装任务，控制并发数
async def limited_task(semaphore, n):
    async with semaphore:
        return await counter(n)

async def main():
    max_concurrent = 4  # 同时运行的任务数
    total_task_num = 20  # 总任务数量，可根据实际需求调整
    semaphore = asyncio.Semaphore(max_concurrent)
    pending_tasks = set()

    # 启动初始的4个任务
    for _ in range(max_concurrent):
        if total_task_num <= 0:
            break
        n = random.randint(100, 500)
        task = asyncio.create_task(limited_task(semaphore, n))
        pending_tasks.add(task)
        total_task_num -= 1

    # 动态调度任务：完成一个补一个
    while pending_tasks:
        # 等待任意一个任务完成
        done_tasks, pending_tasks = await asyncio.wait(
            pending_tasks,
            return_when=asyncio.FIRST_COMPLETED
        )
        # 处理完成的任务（可根据需求添加结果处理逻辑）
        for task in done_tasks:
            result = await task
            print(f'Task with param {result} completed')
        
        # 补充新任务，直到达到并发上限或无任务可补
        while len(pending_tasks) < max_concurrent and total_task_num > 0:
            n = random.randint(100, 500)
            new_task = asyncio.create_task(limited_task(semaphore, n))
            pending_tasks.add(new_task)
            total_task_num -= 1

if __name__ == "__main__":
    asyncio.run(main())

适配实际求解器场景（subprocess调用） #

如果要替换为实际的求解器调用，只需修改counter协程为subprocess异步执行逻辑即可：

async def run_solver(core_count, task_params):
    print(f'Starting solver with {core_count} cores, params: {task_params}')
    # 异步启动求解器进程
    proc = await asyncio.create_subprocess_exec(
        'your_solver_path',  # 替换为你的求解器可执行文件路径
        f'--cores={core_count}',  # 传递核数参数
        *task_params,  # 传递任务的其他参数
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    # 等待进程完成并获取输出
    stdout, stderr = await proc.communicate()
    if proc.returncode == 0:
        print(f'Solver finished successfully for params: {task_params}')
    else:
        print(f'Solver failed for params: {task_params}, code: {proc.returncode}')
        print(f'Error log: {stderr.decode("utf-8")}')
    return task_params

# 对应的limited_task修改为：
async def limited_solver(semaphore, core_count, task_params):
    async with semaphore:
        return await run_solver(core_count, task_params)

在main函数中，启动任务时传递10核参数即可：

new_task = asyncio.create_task(limited_solver(semaphore, 10, your_task_params))

内容的提问来源于stack exchange，提问作者user20426821