锐英源精品原创，禁止全文或局部转载，禁止任何形式的非法使用，侵权必究。点名“简易百科”和闲暇巴盗用锐英源原创内容。

以后台线程模式使用MSMQ微软消息队列

背景

最近开发平台，需要进程间通信，找了些例子，MSMQ微软消息队列的文章比较多，例子也比较多，但是打开本文例子代码运行有错误，如下图：

报异常错误为：此计算机上尚未安装消息。

根据错误理解，要安装消息，找到配置界面如下图：

看到这么复杂，我也就没有用了的心思。因为如果用户偶然卸载了MSMQ，则软件功能受影响，这有点接受不了。

下面是翻译正文，请记住：看不懂codeproject，请找锐英源软件。

 

介绍 

本文介绍了通过System.Messaging命名空间公开的 MSMQ 功能（安装 MSMQ、创建消息队列、消息和推送、从 MSMQ 弹出消息）以及BackgroundWorker功能（如何启动线程并将参数传递给BackgroundWorker、跟踪进度和取消运行时的后台进程）。

为什么选择 MSMQ（Microsoft 消息队列服务器）？

消息队列用于我们需要故障安全机制而两个进程相互通信的场景。Microsoft 提供了用于实现消息队列的 MSMQ。MSMQ 本质上是一种消息传递协议，它允许在不同服务器上运行的应用程序以故障安全方式进行通信。队列是一种临时存储，一个进程可以存储消息，另一个进程可以检索这些消息。这可确保即使系统在一段时间内未连接，消息也不会丢失。

先决条件：设置 MSMQ

确保您有 Windows CD。只需打开控制面板，“添加/删除 Windows 组件”，并确保选中消息队列服务复选框。消息队列必须安装在发送（客户端）和接收机器（服务器）上。

 

完成此步骤后，您将能够创建私人消息队列。专用队列用于“工作组安装”，这意味着计算机未集成到 Active Directory 中。私有队列中存在一些限制——它们不发布，而公共队列是。

创建队列

打开 Microsoft 管理控制台，添加/删除“计算机管理”管理单元，浏览到“服务和应用程序”，展开并选择“消息队列”节点。右键单击“Private Queues”并创建一个新队列：

 

下表中提到了不同类型的队列：

队列类型	描述
公共队列	在目录服务中注册，可以被任何消息队列应用程序定位。
私人队列	在本地机器上注册，通常无法被其他应用程序定位。
系统队列	这些是系统级队列。

在每个队列下，您会发现三个项目：

• 队列消息：显示当前位于队列中的所有消息。您可以查看每条消息的属性，但不能更改它。您不能创建新消息或删除单个消息。您可以通过右键单击“队列消息”并选择“所有任务|清除”来删除所有消息。
• 日志消息：创建队列时，您还可以启用“日志”。这意味着当从该队列接收/读取消息时，副本将被放置到该队列的日志中。因此，您拥有从该队列读取和处理的每条消息的副本。
• 触发器：允许您注册一个触发器，以便在将消息放入此消息队列时调用。设置触发器时，您可以指定是调用 COM 组件还是外部可执行文件。

MSMQ的使用

 

 

对于这个示例，我只使用了一个将数据推送到 MSMQ 并从 MSMQ 弹出数据的应用程序。为此，我使用了两个BackgroundWorker线程。一个线程用于为 MSMQ 创建和推送消息，另一个线程用于弹出和显示队列中的数据。

有几个简单的步骤可以开始使用 MSMQ。添加对System.Messaging解决方案的引用。

 

 

然后创建MessageQueue（提供对消息队列服务器上的队列的访问）、 Message（提供对定义消息队列消息所需的属性的访问）对象。

static System.Messaging.MessageQueue objMessageQueue;
static System.Messaging.Message objMqMessage = new System.Messaging.Message();

该BackgroundWorker组件为您提供了一种在单独的线程上运行耗时任务的方法。实际上，它的工作方式与异步委托相同，但在异步委托方法中，您必须考虑有关使用 UI 元素的一些问题，因为它们在另一个线程上运行。在BackgroundWorker中，编组问题通过基于事件的模型抽象出来。

对于线程，使用工具箱控件中的BackgroundWorker一个线程 ( )：bgwRandomData

 

 

另一个是在运行时（bgwPopMSMQ）创建的（在代码中）：

private BackgroundWorker bgwPopMSMQ;
//Not drag droped from the toolbox, created in runtime.

public Form1()
{
    InitializeComponent();

    this.bgwPopMSMQ = new System.ComponentModel.BackgroundWorker();
    //To Stop the thread the thread should
    //be provided with  WorkerSupportsCancellation as true;
    bgwPopMSMQ.WorkerSupportsCancellation = true;
    this.bgwPopMSMQ.DoWork += new DoWorkEventHandler(bgwPopMSMQ_DoWork);
    this.bgwPopMSMQ.RunWorkerCompleted += 
      new RunWorkerCompletedEventHandler(bgwPopMSMQ_RunWorkerCompleted);
}

使用 BackgroundWorker

对于BackgroundWorker从工具箱中删除的线程（线程 1）重命名为 bgwRandomData，并WorkerReportsProgress 更改为“True”以提供报告进度更新，WorkerSupportsCancellation更改为“True”以取消异步操作。

 

现在下一步是创建这些方法：

• DoWork：这是负责处理大型操作的主要方法。
• ProgressChanged：该方法报告后台workerDoWork方法执行的操作的进度变化。
• RunWorkerCompleted：此方法检查RunWorkerCompletedEventArgs并相应地执行操作。

 

BackgroundWorker线程“ ”bgwRandomData生成一个随机数据字符串（有 10 个数字）并将消息（带有标签“MYDATA”）推送到myqueuetest队列并为文本框显示相同的数据， Graph.

private void bgwRandomData_DoWork(object sender, DoWorkEventArgs e)
{
    Random rnd = new Random();
    while (true)
    {
        string strMessage = string.Empty;
        for (int i = 0; i < 10; i++)
        {
            strMessage = strMessage + (rnd.Next(0, 10)).ToString() + ",";
        }
        PushIntoQueue(strMessage);//Pushing the number to MyTestQueue
        bgwRandomData.ReportProgress(0, (object)strMessage);
        //provides the message for the Testbox, Graph

        System.Threading.Thread.Sleep(1000);//waiting for 1 Sec
        if (bgwRandomData.CancellationPending)//checking for the cancellation
        {
            e.Cancel = true;
            break;
        }
    }
}

private void bgwRandomData_ProgressChanged(object sender, ProgressChangedEventArgs e)
{
    int a = e.ProgressPercentage;//for this application dont consider the progress percentage
    string strMessage = e.UserState.ToString();
    DrawTheGraph(ref pictureBox1, strMessage);
    textBox1.Text = strMessage;
}

当bgwRandomData线程停止时，Runworkercompleted调用该方法：

private void bgwRandomData_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
{
    MessageBox.Show("BackGroundWorker Thread of Pushing is stopped");
}

这是为 MSMQ 推送消息的代码：

void PushIntoQueue(string strMessage)
{
    //Checking whetere the MyTestQueue exists, if not create it.
    if (MessageQueue.Exists(@".\Private$\MyQueueTest"))
        objMessageQueue = new System.Messaging.MessageQueue(@".\Private$\MyQueueTest");
    else
        objMessageQueue = MessageQueue.Create(@".\Private$\MyQueueTest");

    objMqMessage.Body = strMessage;// Assigning the value to message
    objMqMessage.Label = "MYDATA";// Assigning the Label to message
    if (objMqMessage != null)
    {
        objMessageQueue.Send(objMqMessage);//Storeing the value in //Queue
    }
}

在此，首先检查队列是否存在，如果队列不存在，则创建队列，否则将现有队列分配给objMessageQueue（队列对象）。接下来，将字符串分配给消息正文，并将标签更新为“MYDATA”，然后发送给 MSMQ。

 

 

BackgroundWorker线程bgwPopMSMQ从队列中弹出数据并显示为文本框、图形。对于此线程， ProgressChanged 不会（有意）处理事件。因此，当尝试直接用文本更新文本框时，会观察到跨线程操作。为消除跨线程操作，将文本框相关代码修改为：

//For cross thread operations
textBox2.Invoke((MethodInvoker)delegate()
{
     textBox2.Text = m_strmsg;
});
//end For cross thread operations

在BackgroundWorker线程中，如果需要更新任何 UI 元素，更好的方法是使用ProgressChanged事件。与异步委托相比，线程的优势在于BackgroundWorker编组问题通过基于事件的模型抽象出来。

void bgwPopMSMQ_DoWork(object sender, DoWorkEventArgs e)
{
    MessageQueue objMessageQueue = null;
    System.Messaging.Message objMessage = new System.Messaging.Message();
    string m_strmsg = string.Empty;

    //if Queue exists 
    if (MessageQueue.Exists(@".\Private$\MyQueueTest"))
    {
        objMessageQueue = new System.Messaging.MessageQueue(@".\Private$\MyQueueTest");
        objMessageQueue.SetPermissions("Everyone", MessageQueueAccessRights.FullControl);
    }
    else// else create the Queue
    {
        objMessageQueue = MessageQueue.Create(@".\Private$\MyQueueTest");
         objMessageQueue.SetPermissions("Everyone", MessageQueueAccessRights.FullControl);
    }

    // Purge all the Old data
    objMessageQueue.Purge();

    while (true)
    {
        try
        {
            {
                byte[] bt = new byte[10];
                objMessage = objMessageQueue.Receive();//Receive the first message available in the Queue
                if (objMessage.Label.ToUpper() == "MYDATA")//checks for the lable with name "MYDATA"
                {
                    //sets the formatter used to deserialize an object from the message body
                    objMessage.Formatter = new XmlMessageFormatter(new String[] { "System.String,mscorlib" });
                    m_strmsg = objMessage.Body.ToString();
                    DrawTheGraph(ref pictureBox2, m_strmsg);
                    //For cross thread operations
                    textBox2.Invoke((MethodInvoker)delegate()
                    {
                        textBox2.Text = m_strmsg;
                    });
                    //end For cross thread operations
                    }

            }
            if (bgwPopMSMQ.CancellationPending)//checking for the cancellation
            {
                e.Cancel = true;
                break;
            }
        }
        catch (Exception Ex)
        {
        }
    }
}

两个BackgroundWorkers运行时，第一个线程将数据推送给MSMQ，同时将数据显示给textbox1，第二个线程从MSMQ弹出数据，显示给textbox2。