python aioamqp分布式_带aioamqp的异步RabbitMQ使用者

我想出了一个办法。我不确定这是否是最佳实践，但它正在实现我所需要的。在

而不是做“工作”(在这种情况下，异步睡眠)在回调函数中，我在循环中创建一个新任务，并调度一个单独的co例程来运行do\u work()。大概这是有效的，因为它释放了callback()以立即返回。在

我在Rabbit中加载了几百个不同睡眠计时器的事件，当按下面的代码打印时，它们被交错排列。所以它似乎起作用了。希望这对某人有帮助！在@asyncio.coroutine

def do_work(envelope, body):

yield from asyncio.sleep(int(body))

print("consumer {} recved {} ({})".format(envelope.consumer_tag, body, envelope.delivery_tag))

@asyncio.coroutine

def callback(body, envelope, properties):

loop = asyncio.get_event_loop()

loop.create_task(do_work(envelope, body))

@asyncio.coroutine

def receive_log():

try:

transport, protocol = yield from aioamqp.connect('localhost', 5672, login="login", password="password")

except:

print("closed connections")

return

channel = yield from protocol.channel()

exchange_name = 'cloudstack-events'

exchange_name = 'test-async-exchange'

queue_name = 'async-queue-%s' % random.randint(0, 10000)

yield from channel.exchange(exchange_name, 'topic', auto_delete=True, passive=False, durable=False)

yield from asyncio.wait_for(channel.queue(queue_name, durable=False, auto_delete=True), timeout=10)

binding_keys = ['mykey']

for binding_key in binding_keys:

print("binding", binding_key)

yield from asyncio.wait_for(channel.queue_bind(exchange_name=exchange_name,

queue_name=queue_name,

routing_key=binding_key), timeout=10)

print(' [*] Waiting for logs. To exit press CTRL+C')

yield from channel.basic_consume(queue_name, callback=callback)

loop = asyncio.get_event_loop()

loop.create_task(receive_log())

loop.run_forever()