RabbitMQ BasicAck使下一条消息UnAck

场景如下

一开始，准备队列:2UnAcked: 0

一旦consumer.Queue.Dequeue(1000, out bdea);运行，

ReadyQueue: 1UnAcked: 1这是很明显的，我们读到一条消息，但还没有被承认。

问题是当channel.BasicAck(bdea.DeliveryTag, false);运行时，

ReadyQueue: 0UnAcked: 1

处于Ready状态的消息变为unackand ReadyQueue变为"0" !!

现在，在while循环中，当我们使用consumer.Queue.Dequeue(1000, out bdea);查找第二条消息时，bdea返回null，因为Ready状态中没有任何内容。

这就是问题所在，当有Ack发生时，它总是将消息从就绪队列拖到UnAck。因此，下次我丢失此UnAcked消息时，该消息从未从队列中取出。

但是如果我停止进程(控制台应用程序)，UnAck消息返回到就绪状态。

假设开始时有10条消息处于Ready状态，最后它将只处理5条处于unackstate的消息。每次Ack都会使下一条消息UnAck。如果我停止并再次运行(5条消息处于就绪状态)，猜猜会发生什么，3条消息将被处理，2条消息将被unack。(Dequeue只选择一半的消息)

这是我的代码(代码只有RabbitMQ功能，问题是如果你也尝试这个代码)，

public class TestMessages
{
    private ConnectionFactory factory = new ConnectionFactory();
    string billingFileId = string.Empty;
    private IConnection connection = null;
    private IModel channel = null;
    public void Listen()
    {
        try
        {
            #region CONNECT
            factory.AutomaticRecoveryEnabled = true;
            factory.UserName = ConfigurationManager.AppSettings["MQUserName"];
            factory.Password = ConfigurationManager.AppSettings["MQPassword"];
            factory.VirtualHost = ConfigurationManager.AppSettings["MQVirtualHost"];
            factory.HostName = ConfigurationManager.AppSettings["MQHostName"];
            factory.Port = Convert.ToInt32(ConfigurationManager.AppSettings["MQPort"]);
            #endregion

            RabbitMQ.Client.Events.BasicDeliverEventArgs bdea;
            using (connection = factory.CreateConnection())
            {
                string jobId = string.Empty;
                using (IModel channel = connection.CreateModel())
                {
                    while (true) //KEEP LISTNING
                    {
                        if (!channel.IsOpen)
                            throw new Exception("Channel is closed"); //Exit the loop.
                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                        //Prefetch 1 message
                        channel.BasicQos(0, 1, false);
                        String consumerTag = channel.BasicConsume(ConfigurationManager.AppSettings["MQQueueName"], false, consumer);
                        try
                        {
                            //Pull out the message
                            consumer.Queue.Dequeue(1000, out bdea);
                            if (bdea == null)
                            {
                                //Empty Queue
                            }
                            else
                            {
                                IBasicProperties props = bdea.BasicProperties;
                                byte[] body = bdea.Body;
                                string message = System.Text.Encoding.Default.GetString(bdea.Body);
                                try
                                {
                                    channel.BasicAck(bdea.DeliveryTag, false);
                                    ////Heavy work starts now......
                                }
                                catch (Exception ex)
                                {
                                    //Log
                                }
                            }
                        }
                        catch (Exception ex)
                        {
                            //Log it
                        }
                    }
                }
            }
        }
        catch (Exception ex)
        {
            WriteLog.Error(ex);
        }
        finally
        {
            //CleanUp();
        }
    }
}

我错过了什么吗?

public void SubscribeListner()
{
    Subscription subscription = null;
    const string uploaderExchange = "myQueueExchange";
    string queueName = "myQueue";
    while (true)
    {
        try
        {
            if (subscription == null)
            {
                try
                {
                   //CONNECT Code
                    //try to open connection
                    connection = factory.CreateConnection();
                }
                catch (BrokerUnreachableException ex)
                {
                    //You probably want to log the error and cancel after N tries, 
                    //otherwise start the loop over to try to connect again after a second or so.
                    //log.Error(ex);
                    continue;
                }
                
                //crate chanel
                channel = connection.CreateModel();
                // This instructs the channel not to prefetch more than one message
                channel.BasicQos(0, 1, false);
                // Create a new, durable exchange
                channel.ExchangeDeclare(uploaderExchange, ExchangeType.Direct, true, false, null);
                // Create a new, durable queue
                channel.QueueDeclare(queueName, true, false, false, null);
                // Bind the queue to the exchange
                channel.QueueBind(queueName, uploaderExchange, queueName);
                //create subscription
                subscription = new Subscription(channel, uploaderExchange, false);
            }
            BasicDeliverEventArgs eventArgs;
            var gotMessage = subscription.Next(250, out eventArgs);//250 millisecond
            if (gotMessage)
            {
                if (eventArgs == null)
                {
                    //This means the connection is closed.
                    //DisposeAllConnectionObjects();
                    continue;//move to new iterate
                }
                //process message
                subscription.Ack(); 
                //channel.BasicAck(eventArgs.DeliveryTag, false);
            }
        }
        catch (OperationInterruptedException ex)
        {
            //log.Error(ex);
            //DisposeAllConnectionObjects();
        }
        catch(Exception ex)
        {
        }
    }
}