c# - RabbitMQ 消息处于 Unacked 状态

Question

尝试侦听从 RMQ 接收到的消息，消息被接收，但队列中的消息仍处于未确认状态。

           ConnectionFactory factory = new ConnectionFactory();// assign factory details

           IConnection _connection = factory.CreateConnection();
           IModel  _model = _connection.CreateModel();
           _model.ExchangeDeclare("RmqExchangeName", "topic", true);
           var consumer = new EventingBasicConsumer(_model);

            consumer.Received += (object ch, BasicDeliverEventArgs ea) =>
            {
                   var message = Encoding.Default.GetString(ea.Body);

                    MemoryStream payloadstream = new MemoryStream(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(new UriDto { Url = message })));
                    Helper.Log("Start  Message");
                    // Do something with payloadstream
                    Helper.Log("Sent  Message");
                    _model.BasicAck(ea.DeliveryTag, false);
             };
           _model.BasicConsume(RmqQueueName, false, consumer);

标签： c#rabbitmq

Answer 1

您需要将交换绑定到队列才能使用您的消息。

       ConnectionFactory factory = new ConnectionFactory();// assign factory details
       IConnection _connection = factory.CreateConnection();
       IModel  _model = _connection.CreateModel();
       _model.ExchangeDeclare("RmqExchangeName", "topic", true);
       _model.QueueBind(queue: queueName,exchange: "RmqExchangeName",routingKey: "");
       var consumer = new EventingBasicConsumer(_model);

        consumer.Received += (object ch, BasicDeliverEventArgs ea) =>
        {
               var message = Encoding.Default.GetString(ea.Body);

                MemoryStream payloadstream = new MemoryStream(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(new UriDto { Url = message })));
                Helper.Log("Start  Message");
                // Do something with payloadstream
                Helper.Log("Sent  Message");
                _model.BasicAck(ea.DeliveryTag, false);
         };
       _model.BasicConsume(RmqQueueName, true, consumer);