import asyncio
# 消费者
async def consumer(n, q):
print("consumer {}: 进入商店".format(n))
while True:
item = await q.get()
print("consumer {}: 购买产品 {}".format(n, item))
await asyncio.sleep(0.15)
q.task_done()
print("consumer {}: ending".format(n))
# 生产者 A
async def producer_a(q, num_workers):
print("生产者 A: 开始生产")
for i in range(num_workers * 2):
await q.put("A " + str(i))
print("生产者 A: 上架产品 A {}".format(i))
await asyncio.sleep(0.01)
# 生产者 B
async def producer_b(q, num_workers):
print("生产者 B: 开始生产")
for i in range(num_workers * 2):
await q.put("B " + str(i))
print("生产者 B: 上架产品 B {}".format(i))
await asyncio.sleep(0.02)
# 任务调度
async def main(num_consumers, num_workers):
q = asyncio.Queue(maxsize=num_consumers)
print("初始化生产者")
prod_a = [asyncio.create_task(producer_a(q, num_workers))]
prod_b = [asyncio.create_task(producer_b(q, num_workers))]
print("初始化消费者")
consumers = [asyncio.create_task(consumer(i, q)) for i in range(num_consumers)]
print("asyncio 调度开始")
await asyncio.gather(*prod_a, *prod_b)
print("生产者 All: ending")
await q.join()
print("消费者 All: ending")
for c in consumers:
c.cancel()
# main(消费者数量, 生产者数量)
asyncio.run(main(3, 2))
输出结果如下:
初始化生产者
初始化消费者
asyncio 调度开始
生产者 A: 开始生产
生产者 A: 上架产品 A 0
生产者 B: 开始生产
生产者 B: 上架产品 B 0
consumer 0: 进入商店
consumer 0: 购买产品 A 0
consumer 1: 进入商店
consumer 1: 购买产品 B 0
consumer 2: 进入商店
生产者 A: 上架产品 A 1
consumer 2: 购买产品 A 1
生产者 B: 上架产品 B 1
生产者 A: 上架产品 A 2
生产者 A: 上架产品 A 3
consumer 0: 购买产品 B 1
consumer 1: 购买产品 A 2
生产者 B: 上架产品 B 2
consumer 2: 购买产品 A 3
生产者 B: 上架产品 B 3
生产者 All: ending
consumer 0: 购买产品 B 2
consumer 1: 购买产品 B 3
消费者 All: ending
我理解 await asyncio.gather(*prod_a, *prod_b) 执行了 prod_a 和 prod_b ,但 consumers 是怎么被触发的呢?
这是一个专为移动设备优化的页面(即为了让你能够在 Google 搜索结果里秒开这个页面),如果你希望参与 V2EX 社区的讨论,你可以继续到 V2EX 上打开本讨论主题的完整版本。
V2EX 是创意工作者们的社区,是一个分享自己正在做的有趣事物、交流想法,可以遇见新朋友甚至新机会的地方。
V2EX is a community of developers, designers and creative people.