.net - 等待单个 RabbitMQ 消息超时

Question

我想向 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)

肯定有更好的方法吗？

Answer 1

这是我最终在 .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() 不支持超时!这个愚蠢的限制(我一直遇到)意味着如果你再也没有收到消息，你的线程将永远卡住。