异步接收MSMQ消息

内容简介：异步接收MSMQ消息

在这部分，我们将使用ThreadPool 和MSMQ 进行消息收发。MSMQ 是一个分布式队列，通过MSMQ 一个应用程序可以异步地与另外一个应用程序通信。

在一个典型的场景中，我们要向维护一个队列的MSMQ 服务器发送消息，MSMQ 发送方与MSMQ 服务器(特定队列)之间创建一个连接并向那个队列发送消息。一个MSMQ 接收器接收由MSMQ发送方发送的消息。MSMQ 接收方需要监听一个特定的队列以接收发送到这个队列上的消息。MSMQ服务器在MSMQ发送方和接收方之间起到了一个中转的作用，但MSMQ发送方不知道还有一个MSMQ接收方，反之亦然。

在我们的程序中，我们将开发一个MSMQ发送程序(Windows 窗体应用)和一个MSMQ接收程序(控制台应用)。在我们的MSMQ发送方应用程序中(MSMQUI.cs), 我们使用ThreadPool 向队列中发送一条消息：

private void SendButton_Click(object sender, EventArgs e)
{
    int count = Convert.ToInt32(countTxt.Text);
    while (count > 0)
    {
        ThreadPool.QueueUserWorkItem(new WaitCallback(SendMessage), countTxt.Text);
        count--;
    }
}

在上面的代码片段中(MSMQUI.cs), 我们创建了一个计数器来向MSMQ发送消息。我们通过一个静态对象将消息传递给ThreadPool 并传递给WaitCallback 委托一个SendMessage() 方法的引用。换句话说，SendMessage() 方法将会给MSMQ 发送消息。

private void SendMessage(object state)
{
    if (mq != null)
    {
        try
        {
            msg.Body = state.ToString();
            mq.Send(msg);
        }
        catch (InvalidCastException ex)
        {
        }
    }
}

SendMessage() 方法把静态对象转成字符串并发送给MSMQ。

MSMQUI 应用程序运行起来与如下截图类似：

在MSMQ 监听器(MSMQListener.cs)中, 当一条消息到达队列中以后我们从MSMQ服务端异步地接收消息通知。由于这个原因，我们创建了MSMQ 事件处理器并把MessageReceived() 方法名传递给ReceiveCompletedEventHandler 委托。

mq1 = new MessageQueue(@".\private$\myfirstq3");
mq1.ReceiveCompleted += new ReceiveCompletedEventHandler(MessageReceived);

为了开始从MSMQ 接收消息，我们需要在消息队列对象上调用BeginReceive() 方法。

mq1.BeginReceive(new TimeSpan(0,0,2));

BeginReceive() 方法有一个TimeSpan 参数，它用来指示监听器在消息到达队列之前应该等待多长时间。如果没有使用TimeSpan参数，应用程序将会阻塞在BeginReceive() 方发处直到消息到达MSMQ。

当一条消息到达队列后，MessageReceived() 方法由MSMQ事件处理器调用：

private static void MessageReceived(object source, ReceiveCompletedEventArgs asyncResult)
{
    bool isReceivedSucceed = true;
    MessageQueue mq = null;

    try
    {
        //Connect to the queue
        mq = (MessageQueue)source;
        Message m = mq.EndReceive(asyncResult.AsyncResult);
        m.Formatter = new System.Messaging.XmlMessageFormatter(new Type[] { typeof(string) });
        //Process the message here.
        Console.WriteLine(m.Body.ToString());
        ThreadPool.QueueUserWorkItem(new WaitCallback(InvokeMDAO), m.Body);
    }
    catch (MessageQueueException)
    {
        isReceivedSucceed = false;
    }
    catch (Exception ex)
    {
        isReceivedSucceed = false;
        Console.WriteLine(ex.Message);
    }
    finally
    {
        if (isReceivedSucceed)
        {
            mq1.BeginReceive(new TimeSpan(0, 0, 2));
        }
    }
}

为了获取接收到的消息，我们需要在消息队列对象上调用EndReceive() 方法。在我们获取到System.Messaging.Message 对象后，如何处理这个对象就完全取决于我们了。在一个现实世界应用程序中，我们通常会调用System.EnterpriseServices(COM + )对象来进行一些数据库更新操作或者将消息传递给其他对象处理。这里我们可以使用ThreadPool 来保证对象调用可控。我们把消息体放到ThreadPool 中并把InvokeMDAO() 方法名传递给WaitCallback 委托。InvokeMDAO() 方法调用对象上的一个方法并在控制台上打印消息。如之前讨论的那样，我们可以在InvokeMDAO() 方法中调用一个System.EnterpriseServices(COM+)组件。这将直接控制创建的COM+对象数量。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

• 关于在接收POST请求，Tomcat偶发性接收到的参数不全问题排查分析
• 如何突破商品期货Tick接收限制
• SpringMVC接收和响应json数据
• 如何使用 jq 接收 blob 数据
• 转的关于公众号接收信息的返回
• 如何做到每秒接收 100 万个数据包

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。