Search code examples
c#serviceactivemq-classicnmsapache-nms

Consumer service not dequeuing all messages from ActiveMQ queue


I have a windows service that is attempting to consume messages from some activemq queue's. However, it is only getting some of the messages and others are getting stuck in 'messages pending' in the queue. ActiveMQ tells me it has enqueued lets say 500 messages to the consumer but only 300 were dequeued. There is more than one listener being set up in the service. Here's the important part of the code:

private void setupListener(string queue, string brokerUri)
{
    try
    {
        ISession session = connectionConsumers[brokerUri].CreateSession();
        session.CreateConsumer(session.GetQueue(queue))
               .Listener += new MessageListener(consumer_Listener);
    }
    catch (Exception ex)
    {
        Log.Error("An exception has occured setting up listener for " + queue + " on " + brokerUri + ": {0}, {1}", ex, ex.Message);
    }
}

void consumer_Listener(IMessage message)
{
    try
    {
        processLog((message as ITextMessage).Text);
        message.Acknowledge();
    }
    catch (NMSException ex)
    {
        Log.Error("ActiveMQ Connection Failure: {0}, {1}", ex, ex.Message);
    }
    catch (Exception ex)
    {
        Log.Error("An exception has occured trying to process a message: {0}, {1}", ex, ex.Message);
    }
}

Is there something wrong with the way I'm acknowledging messages that would cause certain ones to not be acknowledged? Is it a concurrency issue? I'm not sure if they are all still going through the processLog function (added to my database).

EDIT: I think it has more to do with acknowledgements not happening properly (for some reason). I am not getting exceptions thrown in my logs. However, activemq shows the following: Dispatch Queue is being filled

From what I've read, the dispatch queue is being filled with messages that were sent to the consumer but not acknowledged. Why could this be?


Solution

  • The problem had to do with our queues being virtual destinations.