我正在尝试使用 asyncio/aioamqp 编写一个异步消费者。我的问题是,回调协程(下方)正在阻塞。我将 channel 设置为执行 basic_consume(),并将回调分配为 callback()。回调有一个“yield from asyncio.sleep”语句(模拟“工作”),它从发布者那里获取一个整数并在打印消息之前休眠一段时间。
如果我发布两条消息,一条时间为“10”,紧接着一条时间为“1”,我预计第二条消息会先打印,因为它的休眠时间较短。相反,回调会阻塞 10 秒,打印第一条消息,然后打印第二条消息。
似乎 basic_consume 或回调在某处阻塞。有没有其他方法可以解决这个问题?
@asyncio.coroutine
def callback(body, envelope, properties):
yield from asyncio.sleep(int(body))
print("consumer {} recved {} ({})".format(envelope.consumer_tag, body, envelope.delivery_tag))
@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()
最佳答案
对于那些感兴趣的人,我想出了一个方法来做到这一点。我不确定这是否是最佳做法,但它可以满足我的需要。
我没有在回调中执行“工作”(在本例中为 async.sleep),而是在循环中创建了一个新任务,并安排了一个单独的协程来运行 do_work()。大概这是可行的,因为它释放了 callback() 以立即返回。
我用不同的 sleep 定时器在 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()
关于python - 使用 aioamqp 的异步 RabbitMQ 消费者,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31324470/