C# RabbitMQ 等待一条消息指定超时？

Question

RabbitMQ Wait for a message with a timeout 和 Wait for a single RabbitMQ message with a timeout 中的解决方案似乎不起作用，因为官方 C# 库中没有下一个交付方法，并且 QueueingBasicConsumer 已被弃用，所以它只是到处抛出 NotSupportedException。

如何等待队列中的单个消息达到指定的超时时间？

附言

可以通过 Basic.Get() 来完成，是的，但是，在指定的时间间隔（流量过多，CPU 过多）拉取消息是不好的解决方案。

更新

EventingBasicConsumer 通过实现不支持立即取消。即使您在某个时候调用 BasicCancel，即使您通过 BasicQos 指定预取 - 它仍然会在 Frames 中获取，并且这些帧可以包含多个消息。因此，它不适合单任务执行。不要打扰 - 它只是不适用于单个消息。

Answer 1

有很多方法可以做到这一点。例如，您可以将EventingBasicConsumer 与ManualResetEvent 一起使用，如下所示（这仅用于演示目的 - 最好使用以下方法之一）：

var factory = new ConnectionFactory();
using (var connection = factory.CreateConnection()) {
    using (var channel = connection.CreateModel()) {
        // setup signal
        using (var signal = new ManualResetEvent(false)) {
            var consumer = new EventingBasicConsumer(channel);
            byte[] messageBody = null;                        
            consumer.Received += (sender, args) => {
                messageBody = args.Body;
                // process your message or store for later
                // set signal
                signal.Set();
            };               
            // start consuming
            channel.BasicConsume("your.queue", false, consumer);
            // wait until message is received or timeout reached
            bool timeout = !signal.WaitOne(TimeSpan.FromSeconds(10));
            // cancel subscription
            channel.BasicCancel(consumer.ConsumerTag);
            if (timeout) {
                // timeout reached - do what you need in this case
                throw new Exception("timeout");
            }

            // at this point messageBody is received
        }
    }
}

正如您在 cmets 中所述 - 如果您希望在同一个队列中有多个消息，这不是最好的方法。好吧，无论如何这都不是最好的方法，我包含它只是为了演示ManualResetEvent 的使用，以防库本身不提供超时支持。

如果您正在执行 RPC（远程过程调用，请求-回复） - 您可以在服务器端使用 SimpleRpcClient 和 SimpleRpcServer。客户端将如下所示：

var client = new SimpleRpcClient(channel, "your.queue");
client.TimeoutMilliseconds = 10 * 1000;
client.TimedOut += (sender, args) => {
    // do something on timeout
};                    
var reply = client.Call(myMessage); // will return reply or null if timeout reached

更简单的方法：使用基本的Subscription 类（它在内部使用相同的EventingBasicConsumer，但支持超时，因此您无需自己实现），如下所示：

var sub = new Subscription(channel, "your.queue");
BasicDeliverEventArgs reply;
if (!sub.Next(10 * 1000, out reply)) {
     // timeout
}