.NET 异步 MSMQ答案

【问题标题】：.NET asynchronous MSMQ.NET 异步 MSMQ
【发布时间】：2011-07-26 01:21:44
【问题描述】：

我不明白这是哪里出了问题。基本上，我有一个从消息队列接收并处理消息的程序。程序可以随时停止，在这种情况下，消息循环在程序退出之前完成了它正在执行的操作。我正在尝试使用以下代码来完成此操作：

private MessageQueue q;
private ManualResetEventSlim idle;

public void Start()
{
    idle = new ManualResetEventSlim();
    q.ReceiveCompleted += this.MessageQueue_ReceiveCompleted;    
    q.BeginReceive();
}    

public void Stop()
{ 
    this.q.Dispose();
    this.idle.Wait();    
}

private void MessageQueue_ReceiveCompleted(object sender, 
    ReceiveCompletedEventArgs e)
{
    Message inMsg;
    try
    {
        inMsg = e.Message;
    }
    catch (Exception ex)
    {
        this.idle.Set();
        return;
    }

    // Handle message

    this.q.BeginReceive();
}

正如希望显而易见的那样，Stop 方法处理消息队列，然后等待设置空闲等待句柄（这应该发生，因为在处理时将调用 ReceiveCompleted 事件，但 e.Message 属性应该除外）。

但是，消息循环仍在继续！我已经处理了消息队列，但它仍然设法从中读取，并且没有调用异常处理程序，这意味着 idle.Wait 行将永远等待。

我的理解是，处理消息队列应该结束任何挂起的接收并调用事件，但 e.Message（或 q.EndReceive）应该抛出异常。不是这样吗？如果没有，我还能如何安全地退出我的消息循环？

谢谢

更新：

这是一个完整的例子（假设队列存在）

class Program
{
    static MessageQueue mq;
    static ManualResetEventSlim idleWH;

    static void Main(string[] args)
    {
        idleWH = new ManualResetEventSlim();

        Console.WriteLine("Opening...");
        using (mq = new MessageQueue(@".\private$\test"))
        {
            mq.Formatter = new XmlMessageFormatter(new Type[] { typeof(int) });
            mq.ReceiveCompleted += mq_ReceiveCompleted;

            for (int i = 0; i < 10000; ++i)
                mq.Send(i);

            Console.WriteLine("Begin Receive...");
            mq.BeginReceive();

            Console.WriteLine("Press ENTER to exit loop");
            Console.ReadLine();

            Console.WriteLine("Closing...");

            mq.Close();
        }

        Console.WriteLine("Waiting...");
        idleWH.Wait();

        Console.WriteLine("Press ENTER (ex)");
        //Console.ReadLine();
    }

    static void mq_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
    {
        try
        {
            var msg = mq.EndReceive(e.AsyncResult);
            Console.Title = msg.Body.ToString();

            // Receive next message
            mq.BeginReceive();
        }
        catch (Exception ex)
        {
            idleWH.Set();
            return;
        }
    }
}

【问题讨论】：

只是好奇，为什么不公开WCF service that uses MSMQ as the binding？这将使您不必担心队列的管理和消息的读取，并更多地关注您的逻辑。
这是一个竞争条件。 mq_ReceiveCompleted 与 Dispose() 同时执行，因此您可以接收在调用 BeginReceive() 时未决的消息，或者可能在此之后但在 Dispose() 之前到达的任何消息完全的。您不能依赖两个同时执行的操作的顺序。

标签： c# .net asynchronous msmq beginreceive

【解决方案1】：

完全不确定您是如何完成这项工作的。您必须在事件中调用 MessageQueue.EndReceive()。只有该方法可以抛出异常。查看 ReceiveCompleted 事件的 MSDN 示例代码。并且不要捕获异常，这只会导致无法诊断的故障。捕获您在处置队列时遇到的特定异常，ObjectDisposedException。

【讨论】：

无论如何，它确实有效。据我了解，Message 属性与 q.EndReceive(e.AsyncResult) 的作用相同，实际上，如果我输入 EndReceive 方法，我会得到相同的结果 - 队列在处理后将继续被消耗.
没错，Message 属性 getter 确实是第一次调用 EndReceive。刮那个。抓住稻草，我看不到 MessageQueue 构造函数。队列真的打开了吗？
好的，示例在第一篇文章中。奇怪的是，如果队列是空的（并且不存在填充一些测试数据的循环），那么该示例将按预期工作 - 按下 ENTER 会导致事件处理程序异常并且程序正确退出。它似乎只有在从队列中收到任何消息时才会发生。

【解决方案2】：

    private static volatile bool _shouldStop = false;

。 . .

            _shouldStop = true;
            mq.Close();

。 . .

        try
        {
            var msg = mq.EndReceive(e.AsyncResult);

            if ( _shouldStop)
            {
                idleWH.Set();
                return;
            }

            mq.BeginReceive();
        }

。 . .

【讨论】：

【解决方案3】：

我可以让这个工作的唯一方法是使用事务队列。任何非事务性队列似乎都容易受到此影响。不是答案，而是我能给找到此问题的任何人的最佳建议。

【讨论】：