asyncio之asyncio.Queue 任务队列 生产者-消费者流

asyncio.Queue 示例

import asyncio, random
urls = '''http://m.sohu.com
http://jd.com
http://qq.com
http://sohu.com
http://bilibili.com
http://sina.com.cn
http://youku.com'''
urls = urls.split('\n')

# 包工头 给工人发布任务
async def producer(queue):
    while True:
        # 模拟任务用时
        await asyncio.sleep(random.randint(1,3))
        # 生成一个任务并将其发送给工人
        url = random.choice(urls)
        print(f'包工头：获取一个网址 {url} 写入任务队列中')
        await queue.put(url)


# 工人 接受包工头的任务 开始做任务
async def consumer(queue):
    while True:
        # print(queue.qsize)
        url = await queue.get()
        # 处理从工人接收到的网址
        # 模拟处理任务用时
        await asyncio.sleep(random.randint(1,5))
        queue.task_done()
        print(f'工人完成了任务：{url}')

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

    # 几个包工头，几个工人
    producers = [asyncio.create_task(producer(queue)) for _ in range(3)]
    consumers = [asyncio.create_task(consumer(queue)) for _ in range(10)]

    # 在生产者和消费者都运行的情况下，等待
    # the producers to finish
    await asyncio.gather(*producers)
    print('---- done producing')

    # 等待剩余的任务进行处理
    await queue.join()

    # 取消消费者，它们现在是空闲的
    for c in consumers:
        c.cancel()

asyncio.run(main())