Python 自带异步队列的大坑

青南

发布于 2020-06-04 08:55:25

7.6K00

代码可运行

文章被收录于专栏：未闻Code未闻Code

运行总次数：0

代码可运行

我们在使用 Python 的 asyncio 写异步程序的时候，可能会使用asyncio.Queue来实现一个异步队列，通过它来让生产者和消费者进行通信。

但如果你的异步队列没有填写maxsize参数，那么可能会产生让你意料之外的结果。我们来看一段代码：

import asyncio
import random
import aiohttp


async def producer(queue):
    for _ in range(10):
        sleep_time = random.randint(1, 2)
        await queue.put(sleep_time)


async def consumer(queue):
    while True:
        sleep_time = await queue.get()
        size = queue.qsize()
        print(f'当前队列有：{size} 个元素')
        url = f'http://httpbin.org/delay/{sleep_time}'
        async with aiohttp.ClientSession() as client:
            resp = await client.get(url)
            print(await resp.json())

async def main():
    queue = asyncio.Queue()
    asyncio.create_task(producer(queue))
    con = asyncio.create_task(consumer(queue))
    await con


asyncio.run(main())

这段代码把 producer和consumer分别创建成异步任务，期望实现的效果是生产者不停生产数据放进异步队列，消费者不停从队列读取数据，然后发起网络请求。生产者与消费者利用 IO 等待时间实现并行。

但如果你运行一下这段代码，你会发现一件很奇怪的事情，如下图所示：

当我们的消费者开始消费的时候，队列里面实际上已经有10条数据了！由于图中代码第19行是先读取了一条数据再打印剩余的数量，所以打印的是当前队列有：9 个元素。

所以，生产者与消费者根本没有并行。是生产者里面的循环完全运行完成了，才开始运行的消费者！

如果在实际代码中，你的生产者生产了几百万条数据，那么此时所有数据全部都堆放在异步队列里面，很容易就把你的内存撑爆了！

那么这个问题要如何解决呢？实际上非常简单，使用maxsize参数指定异步队列的大小：

queue = asyncio.Queue(maxsize=3)

我们这里设定为3，再运行看看效果：

看到这里，可能有人会说，这仅仅是生产者先把异步队列堆满，才能进行消费，并没有什么本质区别啊，本质上还是先只有生产者运行，等他跑不动了（队列满了），消费者才能运行，还是没有实现并行啊。

这是由于，在上面的例子中，生产者的速度远远超过消费者的速度，所以才会出现生产者总是堆满队列的问题。

为了说明生产者和消费者能真正利用 IO 等待时间进行并行，我们改一下代码：

import asyncio
import random
import aiohttp


async def producer(queue):
    for i in range(10):
        await queue.put(i)
        await asyncio.sleep(random.randint(1, 3))


async def consumer(queue):
    while True:
        sleep_time = await queue.get()
        size = queue.qsize()
        print(f'当前队列有：{size} 个元素')
        url = 'http://httpbin.org/delay/2'
        async with aiohttp.ClientSession() as client:
            resp = await client.get(url)
            print(await resp.json())

async def main():
    queue = asyncio.Queue(maxsize=3)
    asyncio.create_task(producer(queue))
    con = asyncio.create_task(consumer(queue))
    await con


asyncio.run(main())