Python 处理(潜在)不断增长的任务队列的并行并行方法

IIUC您可以在放置要处理的物品时使用asyncio.Queue，例如:

import asyncio


async def worker(queue: asyncio.Queue):
    while True:
        item = await queue.get()

        if item == 'spawn more jobs':
            print('Spawning more jobs!')
            queue.put_nowait('other1')
            queue.put_nowait('other2')
        else:
            await asyncio.sleep(1)
            print(f'Processed job item: {item}')

        queue.task_done()

async def main():
    q = asyncio.Queue()

    # we have pool of 2 workers that work concurrently:
    workers = [asyncio.create_task(worker(q)) for i in range(2)]

    # initially we have 4 job items (one item spawns 2 more jobs):
    for job in ['job1', 'job2', 'spawn more jobs', 'job3']:
        q.put_nowait(job)

    await q.join()

    for w in workers:
        w.cancel()

asyncio.run(main())

这打印:

Processed job item: job1
Spawning more jobs!
Processed job item: job2
Processed job item: job3
Processed job item: other1
Processed job item: other2