如何订阅 MSMQ 队列但仅在 .Net 中“查看”消息？

Question

我们有一个接收消息并由应用程序处理的 MSMQ 队列设置。我们想让另一个进程订阅队列，然后读取消息并记录它的内容。

我已经有了这个，问题是它一直在偷看队列。运行时服务器上的 CPU 约为 40%。 mqsvc.exe 以 30% 运行，此应用以 10% 运行。我宁愿有一些东西只是等待消息进来，得到它的通知，然后记录它而不不断地轮询服务器。

    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

恕我直言，您应该在队列中打开日志。然后，您可以保证保留已提交到队列的所有消息的副本，而您费力地尝试创建自己的机制来记录所有消息时，情况并非如此。

如果您想要比队列本身更易于阅读的内容（我当然希望如此），那么定期记录和删除日志消息会更容易和更可靠。那么这个过程的运行速度有多快并不重要，您只需要获取一次消息，总体而言，这只是解决问题的更好方法。

Answer 2

您是否尝试过使用MSMQEvent.Arrived 来跟踪消息？

当表示打开队列的 MSMQQueue 对象实例的 MSMQQueue.EnableNotification 方法已被调用并且消息被发现或到达队列中的适用位置时，将触发 MSMQEvent 对象的 Arrived 事件。

Answer 3

这对我有用。它在等待消息时阻塞线程。每个循环周期都会检查类成员 _bServiceRunning 以查看线程是否应该中止。

    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 4

没有任何 API 可以让您只查看每条消息一次。

问题是BeginPeek 会在队列中已有消息时立即执行其回调。由于您没有删除消息（毕竟这是 peek，而不是接收！），当您的回调再次开始 peek 时，该过程重新开始，因此 MessageFound 几乎不断运行。

您最好的选择是将消息记录在写入器或读取器中。 Journaling 将在短期内有效（如果您只关心收到的消息），但不是长期解决方案：

虽然性能开销 从队列中检索消息 仅针对日记功能进行了配置 比检索多出约 20% 没有日记的消息，真正的 成本是意想不到的问题引起的 当未经检查的 MSMQ 服务运行时 内存不足或机器已用完 磁盘空间

Answer 5

在 peek 迭代之间的 Thread.Sleep(10) 可以为您节省大量周期。

我能想到的唯一其他选择是将日志记录构建到队列读取应用程序中。