我想向 RabbitMQ 服务器发送消息,然后等待回复消息(在“回复”队列中)。当然,我不想永远等待,以防处理这些消息的应用程序出现故障 - 需要超时。这听起来像是一项非常基本的任务,但我找不到执行此操作的方法。我现在遇到了这个问题 py-amqplib和 RabbitMQ .NET client .
到目前为止我得到的最好的解决方案是使用 basic_get
和中间的 sleep
进行轮询,但这非常难看:
def _wait_for_message_with_timeout(channel, queue_name, timeout):
slept = 0
sleep_interval = 0.1
while slept < timeout:
reply = channel.basic_get(queue_name)
if reply is not None:
return reply
time.sleep(sleep_interval)
slept += sleep_interval
raise Exception('Timeout (%g seconds) expired while waiting for an MQ response.' % timeout)
肯定有更好的方法吗?
最佳答案
这是我最终在 .NET 客户端中执行的操作:
protected byte[] WaitForMessageWithTimeout(string queueName, int timeoutMs)
{
var consumer = new QueueingBasicConsumer(Channel);
var tag = Channel.BasicConsume(queueName, true, null, consumer);
try
{
object result;
if (!consumer.Queue.Dequeue(timeoutMs, out result))
throw new ApplicationException(string.Format("Timeout ({0} seconds) expired while waiting for an MQ response.", timeoutMs / 1000.0));
return ((BasicDeliverEventArgs)result).Body;
}
finally
{
Channel.BasicCancel(tag);
}
}
不幸的是,我不能对 py-amqplib 做同样的事情,因为它的 basic_consume
方法不会调用回调,除非你调用 channel.wait()
和 channel.wait()
不支持超时!这个愚蠢的限制(我一直遇到)意味着如果你再也没有收到消息,你的线程将永远卡住。
关于.net - 等待单个 RabbitMQ 消息超时,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/2799731/