谁能给个消息队列应用的例子(C#的)

Question

应用程序中主要的类是 ServiceControl，它是从 ServiceBase 继承的。因而，它必须实现 OnStart 和 OnStop 方法，以及可选的 OnPause 和 OnContinue 方法。事实上，类是在静态方法 Main 内构造的：using System;
using System.ServiceProcess;public class ServiceControl: ServiceBase
{
   // 创建服务对象的主入口点
   public static void Main()
   {
      ServiceBase.Run(new ServiceControl());
   }   // 定义服务参数的构造对象
   public ServiceControl()
   {
      CanPauseAndContinue = true;
      ServiceName = "MSDNMessageService";
      AutoLog = false;
   }
   
   protected override void OnStart(string[] args) {...}
   protected override void OnStop() {...}
   protected override void OnPause() {...}
   protected override void OnContinue() {...}
}ServiceControl 类创建一系列 CWorker 对象，即，为需要处理的每个消息队列创建 CWorker 类的一个实例。根据定义中处理队列所需的线程数目，CWorker 类依次创建了一系列的 CWorkerThread 对象。CWorkerThread 类创建的一个处理线程将执行实际的服务工作。使用 CWorker 和 CWorkerThread 类的主要目的是确认服务控件 Start、Stop、Pause 和 Continue 命令。因为这些进程必须是无阻塞的，命令操作最终将在后台处理线程上执行。CWorkerThread 是一个抽象类，被 CWorkerThreadAppSpecific 、CWorkerThreadRoundRobin 和 CWorkerThreadAssembly 继承。这些类以不同的方式处理消息。前两个类通过给另一队列发送消息来处理消息（其不同之处在于确定接收队列路径的方式），最后一个类则使用消息属性来调用组件方法。 .NET 框架内部的错误处理是以基类 Exception 为基础的。当系统引发或捕获错误时，这些错误必须是从 Exception 中导出的类。CWorkerThreadException 类就是这样一种实现，它通过附加额外属性（用于定义服务是否应继续运行）来扩展基类。最后，应用程序包含两种结构。这些值类型定义了辅助进程或线程的运行时参数，以简化 CWorker 和 CWorkerThread 对象的结构。使用值类型结构（而不是引用类型类）能够确保这些运行时参数维护的是数值（而不是引用）。IWebMessage 接口
CWorkerThread 的实现之一是一个调用组件方法的类。这个名为 CWorkerThreadAssembly 的类使用 IWebMessage 接口来定义服务和组件之间的约定。与当前版本的 Microsoft Visual Studio® 不同，C# 接口可以在任何语言中显式定义，而不需要创建和编译 IDL 文件。C# IWebMessage 接口的定义如下：public interface IWebMessage
{
   WebMessageReturn Process(string sMessageLabel, string sMessageBody, int iAppSpecific);
   void Release();
}ATL 代码中的 Process 方法是为处理消息而指定的。Process 方法的返回代码定义为枚举类型 WebMessageReturn：public enum WebMessageReturn 
{
   ReturnGood,
   ReturnBad,
   ReturnAbort
}枚举的定义如下：Good 表示继续处理，Bad 表示将消息写入错误队列，Abort 表示终止处理。Release 方法为服务提供了轻松清除类实例的途径。因为仅在垃圾回收的过程中才调用类实例的析构函数，所以确保所有占用昂贵资源（例如数据库连接）的类都有一个能够在析构之前被调用的方法，用来释放这些资源，这是一种非常好的构思。名称空间
在这里先简单介绍一下名称空间。名称空间允许在内部和外部表示中将应用程序组织成为逻辑元素。服务内的所有代码都包含在 MSDNMessageService.Service 名称空间内。尽管服务代码包含在若干文件中，但是由于它们包含在同一名称空间中，因此用户不需要引用其他文件。由于 IWebMessage 接口包含在 MSDNMessageService.Interface 名称空间中，因此使用此接口的线程类具有一个接口名称空间。服务类
应用程序的目的是监视和处理消息队列，每一队列在收到消息时都执行不同的进程。应用程序是作为 Windows 服务来实现的。ServiceBase 类
如前所述，服务的基本结构是从 ServiceBase 继承的类。重要的方法包括 OnStart、OnStop、OnPause 和 OnContinue，每一个替代方法都与一个服务控制操作直接对应。OnStart 方法的目的是创建 CWorker 对象，而 CWorker 类又创建 CWorkerThread 对象，然后在该对象中创建执行服务工作的线程。服务的运行时配置（以及 CWorker 和 CWorkerThread 对象的属性）是在基于 XML 的配置文件中维护的。它的名称与创建的 .exe 文件相同，但带有一个 .cfg 后缀。配置示例如下：

Answer 1

<?xml version="1.0"?>
<configuration>
<ProcessList>
  <ProcessDefinition
       ProcessName="Worker1"
       ProcessDesc="Message Worker with 2 Threads"
       ProcessType="AppSpecific"
       ProcessThreads="2"
       InputQueue=".\private$\test_load1"
       ErrorQueue=".\private$\test_error">
    <OutputList>
      <OutputDefinition OutputName=".\private$\test_out11" />
      <OutputDefinition OutputName=".\private$\test_out12" />
    </OutputList>
  </ProcessDefinition>
  <ProcessDefinition
       ProcessName="Worker2"
       ProcessDesc="Assembly Worker with 1 Thread"
       ProcessType="Assembly"
       ProcessThreads="1"
       InputQueue=".\private$\test_load2"
       ErrorQueue=".\private$\test_error">
    <OutputList>
      <OutputDefinition OutputName="C:\MSDNMessageService\MessageExample.dll" />
      <OutputDefinition OutputName="MSDNMessageService.MessageSample.ExampleClass"/>
    </OutputList>
  </ProcessDefinition>
</ProcessList>
</configuration>对此信息的访问通过来自 System.Configuration 程序集的 ConfigManager 类来管理。静态 Get 方法返回信息的集合，这些集合将被枚举以获得单个属性。这些属性集的设置决定了辅助对象的运行时特征。除了这一配置文件，您还应该创建定义 XML 文件结构的图元文件，并在其中引用位于服务器 machine.cfg 配置文件中的图元文件：<?xml version ="1.0"?>
<MetaData xmlns="x-schema:CatMeta.xms">
   <DatabaseMeta InternalName="MessageService">
   <ServerWiring Interceptor="Core_XMLInterceptor"/>
   <Collection 
         InternalName="Process" PublicName="ProcessList"
         PublicRowName="ProcessDefinition"
         SchemaGeneratorFlags="EMITXMLSCHEMA">
      <Property InternalName="ProcessName" Type="String" MetaFlags="PRIMARYKEY" />
      <Property InternalName="ProcessDesc" Type="String" />
      <Property InternalName="ProcessType" Type="Int32" DefaultValue="RoundRobin" >
         <Enum InternalName="RoundRobin"  Value="0"/>
         <Enum InternalName="AppSpecific" Value="1"/>
         <Enum InternalName="Assembly" Value="2"/>
      </Property>
      <Property InternalName="ProcessThreads" Type="Int32" DefaultValue="1" />
      <Property InternalName="InputQueue" Type="String" />
      <Property InternalName="ErrorQueue" Type="String" />
      <Property InternalName="OutputName" Type="String" />
      <QueryMeta InternalName="All" MetaFlags="ALL" />
      <QueryMeta InternalName="QueryByFile" CellName="__FILE" Operator="EQUAL"  />
   </Collection>
   <Collection 
         InternalName="Output" PublicName="OutputList"
         PublicRowName="OutputDefinition"
         SchemaGeneratorFlags="EMITXMLSCHEMA">
      <Property InternalName="ProcessName" Type="String" MetaFlags="PRIMARYKEY" />
      <Property InternalName="OutputName" Type="String" MetaFlags="PRIMARYKEY" />
      <QueryMeta InternalName="All" MetaFlags="ALL" />
      <QueryMeta InternalName="QueryByFile" CellName="__FILE" Operator="EQUAL"  />
   </Collection>
   </DatabaseMeta>
   <RelationMeta   
      PrimaryTable="Process" PrimaryColumns="ProcessName"
      ForeignTable="Output"  ForeignColumns="ProcessName"
      MetaFlags="USECONTAINMENT"/>
</MetaData>由于 Service 类必须维护一个已创建辅助对象的列表，因此使用了 Hashtable 集合，用于保持类型对象的名称/数值对列表。Hashtable 不仅支持枚举，还允许通过关键字来查询值。在应用程序中，XML 进程名称是唯一的关键字：private Hashtable htWorkers = new Hashtable();
IConfigCollection cWorkers = ConfigManager.Get("ProcessList", new AppDomainSelector());
foreach (IConfigItem ciWorker in cWorkers)
{
   WorkerFormatter sfWorker = new WorkerFormatter();
   sfWorker.ProcessName = (string)ciWorker["ProcessName"];
   sfWorker.ProcessDesc = (string)ciWorker["ProcessDesc"];
   sfWorker.NumberThreads = (int)ciWorker["ProcessThreads"];
   sfWorker.InputQueue = (string)ciWorker["InputQueue"];
   sfWorker.ErrorQueue = (string)ciWorker["ErrorQueue"];
   // 计算并定义进程类型
   switch ((int)ciWorker["ProcessType"])
   {
      case 0:
         sfWorker.ProcessType = WorkerFormatter.SFProcessType.ProcessRoundRobin;
         break;
      case 1:
         sfWorker.ProcessType = WorkerFormatter.SFProcessType.ProcessAppSpecific;
         break;
      case 2:
         sfWorker.ProcessType = WorkerFormatter.SFProcessType.ProcessAssembly;
         break;
      default:
         throw new Exception("Unknown Processing Type");
   }
   // 执行更多的工作以读取输出信息
   string sProcessName = (string)ciWorker["ProcessName"];
   if (htWorkers.ContainsKey(sProcessName))
      throw new ArgumentException("Process Name Must be Unique: " + sProcessName);
   htWorkers.Add(sProcessName, new CWorker(sfWorker));
}在这段代码中没有包含的主要信息是输出数据的获取。每一个进程定义中都有一组相应的输出定义项。该信息是通过如下的简单查询读取的：string sQuery = "SELECT * FROM OutputList WHERE ProcessName=" + 
   sfWorker.ProcessName + " AND Selector=appdomain://";
ConfigQuery qQuery = new ConfigQuery(sQuery);
IConfigCollection cOutputs = ConfigManager.Get("OutputList", qQuery);
int iSize = cOutputs.Count, iLoop = 0;
sfWorker.OutputName = new string[iSize];
foreach (IConfigItem ciOutput in cOutputs)
   sfWorker.OutputName[iLoop++] = (string)ciOutput["OutputName"];CWorkerThread 和 Cworker 类都有相应的服务控制方法，根据服务控制操作进行调用。由于 Hashtable 中引用了每一个 CWorker 对象，因此需要枚举 Hashtable 的内容，以调用适当的服务控制方法：foreach (CWorker cWorker in htWorkers.Values)
   cWorker.Start();类似地，实现的 OnPause、OnContinue 和 OnStop 方法是通过调用 CWorker 对象上的相应方法来执行操作的。CWorker 类
CWorker 类的主要功能是创建和管理 CWorkerThread 对象。Start、Stop、Pause 和 Continue 方法调用相应的 CWorkerThread 方法。实际的 CWorkerThread 对象是在Start 方法中创建的。与使用 Hashtable 管理辅助对象引用的 Service 类相似，CWorker 使用 ArrayList（简单的动态数组）来维护线程对象的列表。在这个数组内部，CWorker 类创建了 CWorkerThread 类的一个实现版本。CWorkerThread 类（将在下面讨论）是一个必须继承的抽象类。导出类定义了消息的处理方式：aThreads = new ArrayList();
for (int idx=0; idx<sfWorker.NumberThreads; idx++)
{
   WorkerThreadFormatter wfThread = new WorkerThreadFormatter();
   wfThread.ProcessName = sfWorker.ProcessName;
   wfThread.ProcessDesc = sfWorker.ProcessDesc;
   wfThread.ThreadNumber = idx;
   wfThread.InputQueue = sfWorker.InputQueue;
   wfThread.ErrorQueue = sfWorker.ErrorQueue;
   wfThread.OutputName = sfWorker.OutputName;
   // 定义辅助类型，并将其插入辅助线程结构
   CWorkerThread wtBase;
   switch (sfWorker.ProcessType)
   {
      case WorkerFormatter.SFProcessType.ProcessRoundRobin:
         wtBase = new CWorkerThreadRoundRobin(this, wfThread);
         break;
      case WorkerFormatter.SFProcessType.ProcessAppSpecific:
         wtBase = new CWorkerThreadAppSpecific(this, wfThread);
         break;
      case WorkerFormatter.SFProcessType.ProcessAssembly:
         wtBase = new CWorkerThreadAssembly(this, wfThread);
         break;
      default:
         throw new Exception("Unknown Processing Type");
   }
   // 添加对数组的调用
   aThreads.Insert(idx, wtBase);
}

Answer 2

参考：
http://search.csdn.net/expert/topic/50/5001/2002/10/23/1118013.htm
http://www.csdn.net/Develop/Read_Article.asp?Id=13612
http://search.csdn.net/expert/topic/52/5201/2002/10/24/1122478.htmC#：消息队列应用程序
http://www.yesky.com/20011217/210052.shtml
 
怎样用MSMQ
http://www.codeproject.com/csharp/msmqchannel.asp什么是MSMQ
http://www.microsoft.com/msmq/default.htm深入理解基层知识—— COM+和MTS, DCOM和MSMQ,.NET中的序列化
http://www.itraining.net.cn/lzh/showArticle.asp?art_ID=694http://www.csdn.net/develop/read_article.asp?id=19216

Answer 3

.net+msmq快速访问数据库
http://211.147.5.156:7001/pub/article/c1135_a37984_p1.html