1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
import asyncio
import aio_pika
async def process_message(
message: aio_pika.abc.AbstractIncomingMessage,
) -> None:
async with message.process():
print(message.body)
await asyncio.sleep(1)
async def main() -> None:
connection = await aio_pika.connect_robust(
"amqp://guest:guest@127.0.0.1/",
)
queue_name = "test_queue"
# Creating channel
channel = await connection.channel()
# Maximum message count which will be processing at the same time.
await channel.set_qos(prefetch_count=100)
# Declaring queue
queue = await channel.declare_queue(queue_name, auto_delete=True)
await queue.consume(process_message)
try:
# Wait until terminate
await asyncio.Future()
finally:
await connection.close()
if __name__ == "__main__":
asyncio.run(main())
|
