MessageQueue.BeginReceive 是如何工作的以及如何正确使用它？

Question

我目前有一个后台线程。在这个线程中是一个无限循环。

此循环偶尔会更新数据库中的一些值，然后在 MessageQueue 上侦听 1 秒（使用 queue.Receive(TimeSpan.FromSeconds(1)) ）。

只要没有消息进来，这个调用就会在内部抛出一个 MessageQueueException (Timeout) 被捕获，然后循环继续。如果有消息调用正常返回并处理消息，之后循环继续。

这会导致 很多 的第一次机会异常（每秒，除非有要处理的消息），这会在调试输出中产生垃圾邮件，并且当我忘记排除 MessageQueueExceptions 时也会中断调试器。

那么，MessageQueue 的异步处理是如何正确完成的，同时仍然确保，只要我的应用程序运行，队列就会受到监控，并且数据库也会不时更新。当然这里的线程不应该用完100%的CPU。

我只需要大图或一些正确完成异步处理的提示。

Answer 1

我建议不要在线程中循环，而是为 MessageQueue 的 ReceiveCompleted 事件注册一个委托，如下所述：

使用系统； 使用 System.Messaging；

命名空间 MyProject { /// /// 为示例提供一个容器类。 /// 公共类 MyNewQueue {

    //**************************************************
    // Provides an entry point into the application.
    //       
    // This example performs asynchronous receive operation
    // processing.
    //**************************************************

    public static void Main()
    {
        // Create an instance of MessageQueue. Set its formatter.
        MessageQueue myQueue = new MessageQueue(".\\myQueue");
        myQueue.Formatter = new XmlMessageFormatter(new Type[]
            {typeof(String)});

        // Add an event handler for the ReceiveCompleted event.
        myQueue.ReceiveCompleted += new 
            ReceiveCompletedEventHandler(MyReceiveCompleted);

        // Begin the asynchronous receive operation.
        myQueue.BeginReceive();

        // Do other work on the current thread.

        return;
    }


    //**************************************************
    // Provides an event handler for the ReceiveCompleted
    // event.
    //**************************************************

    private static void MyReceiveCompleted(Object source, 
        ReceiveCompletedEventArgs asyncResult)
    {
        // Connect to the queue.
        MessageQueue mq = (MessageQueue)source;

        // End the asynchronous Receive operation.
        Message m = mq.EndReceive(asyncResult.AsyncResult);

        // Display message information on the screen.
        Console.WriteLine("Message: " + (string)m.Body);

        // Restart the asynchronous Receive operation.
        mq.BeginReceive();

        return; 
    }
}

}

来源：https://docs.microsoft.com/en-us/dotnet/api/system.messaging.messagequeue.receivecompleted?view=netframework-4.7.2

Answer 2

您是否考虑过从 MessageQueue.GetMessageEnumerator2 返回的 MessageEnumerator ？

• 您将获得队列的动态内容，以在迭代期间检查和删除队列中的消息。
• 如果没有消息，则 MoveNext() 将返回 false，您无需捕获第一次机会异常
• 如果在您开始迭代后有新消息，那么它们将被迭代（如果它们放在光标之后）。
• 如果光标前有新消息，那么您可以重置迭代器或继续（如果您目前不需要优先级较低的消息）。

Answer 3

与 Jamie Dixon 的评论相反，这种情况非常特殊。注意方法及其参数的命名：BeginReceive(TimeSpan timeout)

如果该方法被命名为 BeginTryReceive，那么如果没有收到任何消息就完全正常了。将其命名为 BeginReceive（或 Receive，对于同步版本）意味着消息应该进入队列。 TimeSpan 参数被命名为 timeout 也很重要，因为超时是异常的。超时意味着预期响应，但没有给出响应，调用者选择停止等待并假设发生了错误。当您使用 1 秒超时调用 BeginReceive/Receive 时，您是在说如果到那时还没有消息进入队列，那么一定是出了问题，我们需要处理它。

如果我理解您想要正确执行的操作，我将执行此操作的方式是：

• 如果我没有将空队列视为错误，则调用 BeginReceive 时超时非常大，或者甚至没有超时。
• 将事件处理程序附加到 ReceiveCompleted 事件，该事件 1) 处理消息，以及 2) 再次调用 BeginReceive。
• 我不会使用无限循环。在使用 BeginReceive 等异步方法时，这既是不好的做法，也是完全多余的。
• 编辑：要放弃没有被任何客户端读取的队列，让队列编写者查看队列以确定它是否“死亡”。

编辑：我有另一个建议。由于我不知道您的应用程序的详细信息，我不知道它是否可行或合适。在我看来，您基本上是在客户端和服务器之间建立连接，消息队列作为通信通道。为什么这是一个“连接”？因为如果没有人在听，队列将不会被写入。我认为，这几乎就是一种联系。使用套接字或命名管道来传输消息不是更合适吗？这样，客户端在完成读取后只需关闭 Stream 对象，并立即通知服务器。正如我所说，我不知道它是否适用于你正在做的事情，但感觉它是一个更合适的沟通渠道。