代码拉取完成,页面将自动刷新
Wrapper for the PIKA for asyncio and humans. See examples and the tutorial in documentation.
If you are newcomer in the RabbitMQ let's start the adopted official RabbitMQ tutorial
pip install aio-pika
Simple consumer:
import asyncio
import aio_pika
async def main(loop):
connection = await aio_pika.connect_robust("amqp://guest:guest@127.0.0.1/", loop=loop)
queue_name = "test_queue"
# Creating channel
channel = await connection.channel() # type: aio_pika.Channel
# Declaring queue
queue = await channel.declare_queue(queue_name, auto_delete=True) # type: aio_pika.Queue
async for message in queue:
with message.process():
print(message.body)
if queue.name in message.body.decode():
break
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
loop.close()
Simple publisher:
import asyncio
import aio_pika
async def main(loop):
connection = await aio_pika.connect_robust("amqp://guest:guest@127.0.0.1/", loop=loop)
routing_key = "test_queue"
channel = await connection.channel() # type: aio_pika.Channel
await channel.default_exchange.publish(
aio_pika.Message(
body='Hello {}'.format(routing_key).encode()
),
routing_key=routing_key
)
await connection.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
loop.close()
Get single message example:
import asyncio
from aio_pika import connect_robust, Message
async def main(loop):
connection = await connect_robust("amqp://guest:guest@127.0.0.1/", loop=loop)
queue_name = "test_queue"
routing_key = "test_queue"
# Creating channel
channel = await connection.channel()
# Declaring exchange
exchange = await channel.declare_exchange('direct', auto_delete=True)
# Declaring queue
queue = await channel.declare_queue(queue_name, auto_delete=True)
# Binding queue
await queue.bind(exchange, routing_key)
await exchange.publish(
Message(
bytes('Hello', 'utf-8'),
content_type='text/plain',
headers={'foo': 'bar'}
),
routing_key
)
# Receiving message
incoming_message = await queue.get(timeout=5)
# Confirm message
incoming_message.ack()
await queue.unbind(exchange, routing_key)
await queue.delete()
await connection.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
See another examples and the tutorial in documentation.
This software follows Semantic Versioning
You feel free to create pull request, but you should describe your cases and add some examples.
The changes should follow simple rules:
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。