How do I subscribe to a MSMQ queue but only “peek” the message in .Net?

Question

We have a MSMQ Queue setup that receives messages and is processed by an application. We'd like to have another process subscribe to the Queue and just read the message and log it's contents.

I have this in place already, the problem is it's constantly peeking the queue. CPU on the server when this is running is around 40%. The mqsvc.exe runs at 30% and this app runs at 10%. I'd rather have something that just waits for a message to come in, get's notified of it, and then logs it without constantly polling the server.

    Dim lastid As String
    Dim objQueue As MessageQueue
    Dim strQueueName As String

    Public Sub Main()
        objQueue = New MessageQueue(strQueueName, QueueAccessMode.SendAndReceive)
        Dim propertyFilter As New MessagePropertyFilter
        propertyFilter.ArrivedTime = True
        propertyFilter.Body = True
        propertyFilter.Id = True
        propertyFilter.LookupId = True
        objQueue.MessageReadPropertyFilter = propertyFilter
        objQueue.Formatter = New ActiveXMessageFormatter
        AddHandler objQueue.PeekCompleted, AddressOf MessageFound

        objQueue.BeginPeek()
    end main

    Public Sub MessageFound(ByVal s As Object, ByVal args As PeekCompletedEventArgs)

        Dim oQueue As MessageQueue
        Dim oMessage As Message

        ' Retrieve the queue from which the message originated
        oQueue = CType(s, MessageQueue)

            oMessage = oQueue.EndPeek(args.AsyncResult)
            If oMessage.LookupId <> lastid Then
                ' Process the message here
                lastid = oMessage.LookupId
                ' let's write it out
                log.write(oMessage)
            End If

        objQueue.BeginPeek()
    End Sub

Answer 1

A Thread.Sleep(10) in between peek iterations may save you a bunch of cycles.

The only other option I can think of is to build the logging into the queue reading application.

Answer 2

Have you tried using MSMQEvent.Arrived to track the messages?

The Arrived event of the MSMQEvent object is fired when the MSMQQueue.EnableNotification method of an instance of the MSMQQueue object representing an open queue has been called and a message is found or arrives at the applicable position in the queue.

Answer 3

There's no API that will let you peek at each message only once.

The problem is that BeginPeek executes its callback immediately if there's already a message on the queue. Since you aren't removing the message (this is peek after all, not receive!), when your callback begins peeking again the process starts over, so MessageFound runs almost constantly.

Your best options are to log the messages in the writer or the reader. Journaling will work for short periods (if you only care about messages that are received), but aren't a long-term solution:

While the performance overhead of retrieving messages from a queue that is configured for Journaling is only about 20% more than retrieving messages without Journaling, the real cost is unexpected problems caused when an unchecked MSMQ service runs out of memory or the machine is out of disk space

Answer 4

This works for me. It blocks the thread while waiting for a message. Each loop cycle checks the class member _bServiceRunning to see if the thread should abort.

    private void ProcessMessageQueue(MessageQueue taskQueue)
    {
        // Set the formatter to indicate body contains a binary message:
        taskQueue.Formatter = new BinaryMessageFormatter();

        // Specify to retrieve selected properties.
        MessagePropertyFilter myFilter = new MessagePropertyFilter();
        myFilter.SetAll();
        taskQueue.MessageReadPropertyFilter = myFilter;

        TimeSpan tsQueueReceiveTimeout = new TimeSpan(0, 0, 10); // 10 seconds

        // Monitor the MSMQ until the service is stopped:
        while (_bServiceRunning)
        {
            rxMessage = null;

            // Listen to the queue for the configured duration:
            try
            {
                // See if a message is available, and if so remove if from the queue if any required
                // web service is available:
                taskQueue.Peek(tsQueueReceiveTimeout);

                // If an IOTimeout was not thrown, there is a message in the queue
                // Get all the messages; this does not remove any messages
                Message[] arrMessages = taskQueue.GetAllMessages();

                // TODO: process the message objects here;
                //       they are copies of the messages in the queue
                //       Note that subsequent calls will return the same messages if they are
                //       still on the queue, so use some structure defined in an outer block
                //       to identify messages already processed.

            }
            catch (MessageQueueException mqe)
            {
                if (mqe.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
                {
                    // The peek message time-out has expired; there are no messages waiting in the queue
                    continue; // at "while (_bServiceRunning)"
                }
                else
                {
                    ErrorNotification.AppLogError("MSMQ Receive Failed for queue: " + mqs.Name, mqe);
                    break; // from "while (_bServiceRunning)"
                }
            }
            catch (Exception ex)
            {
                ErrorNotification.AppLogError("MSMQ Receive Failed for queue: " + mqs.Name, ex);
                break; // from "while (_bServiceRunning)"
            }
        }

    } // ProcessMessageQueue()

Answer 5

IMHO you should just turn on journaling on the queue. Then you are guaranteed a copy is kept of all messages that have been committed to the queue, and that just isn't the case with your laborous attempt to make your own mechanism to log it all.

Much, much easier and more reliable is to log and remove the journaled messages on a scheduled basis if you want something more easily readable than the queue itself (and I certainly would want that). Then it doesn't really matter how fast or not the process is working, you only need to get the messages once, and it's overall just a much better way to solve the problem.