代码之家 › 专栏 › 技术社区 › Bob Horn

如何让我的RabbitMQ消费者只接收一条消息,而不超时并再次发送?

rabbitmq c#

Bob Horn · 技术社区 · 5 年前

我们有一个问题,如果一条消息需要太长时间来处理,它会从未确认返回到就绪状态,并再次被发送。有没有办法防止这种情况发生?

我们有一个队列使用者类,其初始化方式如下:

    private void Initialize()
    {
        string queueName = _configuration["QueueName"] + _configuration["QueueNameSuffixForSubscriber"];

        var factory = CreateConnectionFactory();

        var connection = factory.CreateConnection();
        _channel = connection.CreateModel();

        //Â The message TTL must match for QueueDeclare() to work.
        var arguments = new Dictionary<string, object>();
        arguments.Add("x-message-ttl", Convert.ToInt32(_configuration["EventBusMessageTtl"]));

        //Â ByÂ default,Â RabbitMQÂ dispatchesÂ allÂ theÂ messagesÂ toÂ theÂ firstÂ consumer.Â YouÂ canÂ changeÂ thisÂ behaviourÂ by
        //Â settingÂ theÂ BasicQos,Â thisÂ controlsÂ theÂ noÂ ofÂ messagesÂ aÂ consumerÂ canÂ receiveÂ beforeÂ itÂ acknowledgesÂ it.
        _channel.BasicQos(0, 1, false);

        _channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: arguments);

        var consumer = new EventingBasicConsumer(_channel);
        consumer.Received += ConsumerReceived;

        _channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
    }

注意这一行:

_channel.BasicQos(0, 1, false);

这就是我们的消费者一次只提取一条信息的方式。但是,如果该消息在发送ack之前花费的时间超过2分钟,RMQ将再次发送该消息。我们不想那样。(处理过程几乎不应该超过2分钟,但我们不想满足于几乎 .)

有没有办法阻止RMQ再次发送消息?我可以寄回信之前正在处理消息,但我们会立即收到下一条消息,我们也不希望这样。我们希望在接受下一条消息之前,等待消息处理完成。

如果我们能拉当我们准备好的时候,从RMQ可以解决这个问题。

这是答案 ConsumerReceived() 方法:

    private void ConsumerReceived(object model, BasicDeliverEventArgs eventArgs)
    {
        try
        {
            var message = Encoding.UTF8.GetString(eventArgs.Body.ToArray());
            InvokeHandlers(eventArgs, message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error occurred in processing message or invoking handlers.");
        }
        finally
        {
            _channel.BasicAck(eventArgs.DeliveryTag, false);
        }
    }

0 回复 | 直到 5 年前

David L 5 年前

我同意这似乎是一个理想的流程 polling 而不是消费者订阅。通常情况下,您不想进行投票,因为这会极大地损害吞吐量,但在您的情况下,这正是您想要的。

while (true)
{
    BasicGetResult result = channel.BasicGet(queueName, noAck);
    if (result == null) 
    {
        // No message available at this time.
        // Sleep/delay to avoid cpu and I/O churn
        Thread.Sleep(2000);
    } 
    else 
    {
        try 
        {
            IBasicProperties props = result.BasicProperties;
            var message = Encoding.UTF8.GetString(result.Body.ToArray());
            InvokeHandlers(eventArgs, message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error occurred in processing message or invoking handlers.");
        }
        finally
        {
            _channel.BasicAck(eventArgs.DeliveryTag, false);
        }
    }
}