在Asyncio Python中混淆生产者和消费者

Question

我有一个生产者和消费者异步函数。生产者首先运行，然后三个消费者同时使用该任务。我想无限期地继续这种节奏。

• 停止生产者并向消费者发送信号以开始运行的最佳方法是什么
• 停止使用者并向生产者发送信号以开始运行的最佳方法是什么

我当前的设置如下：

import asyncio
import time

async def producer(event):
n = 0
while True:
print("Running producer...")
await asyncio.sleep(0.5)
n += 1
if n == 2:
event.set()
break

async def consumer(event):
await event.wait()
print("Running consumer...")
await asyncio.sleep(0.5)

async def main():
event = asyncio.Event()
tasks = [asyncio.create_task(producer(event))] + [
asyncio.create_task(consumer(event)) for _ in range(3)
]
await asyncio.gather(*tasks)

while True:
asyncio.run(main())
print("nSleeping for 1 sec...n")
time.sleep(1)

这会产生以下输出：

Running producer...
Running producer...
Running consumer...
Running consumer...
Running consumer...
Sleeping for 1 sec...
Running producer...
Running producer...
Running consumer...
Running consumer...
Running consumer...

上面的代码段将无限期地运行生产者和两个消费者。如预期：

• 生产者和两个消费者同时运行
• 围绕asyncio.run的while循环正在无限期地运行系统

但是，我想知道是否有更好的同步技术来实现这种长时间运行的周期性？

Answer 1

使用事件唤醒消费者的另一种选择是使用队列同步生产者和消费者。代码看起来像

import asyncio

async def producer(queue: asyncio.Queue):
while True:
print("Running producer...")
message = await fetch_message_from_sqs()
if message:
await queue.put(message)
await asyncio.sleep(0.5)

async def consumer(queue: asyncio.Queue):
while True:
print("Running consumer...")
if queue.empty():
await asyncio.sleep(1)
continue
message = await queue.get()
print(message)
await acknowledge_message(message)

async def main():
# You can set a max size if you want to prevent pull too many messages from SQS.
queue = asyncio.Queue()
tasks = asyncio.gather(producer(queue), *[consumer(queue) for _ in range(3)])

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