技术开发 频道

C# 中的消息队列应用程序


  由于 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);
}

一旦所有的对象都已创建,就可以通过调用每个线程对象的 Start 方法来启动它们:

foreach(CWorkerThread cThread in aThreads)
cThread.Start();

Stop、Pause 和 Continue 方法在 foreach 循环里执行的操作类似。Stop 方法具有如下的垃圾收集操作:

GC.SuppressFinalize(this);

  在类析构函数中将调用 Stop 方法,这样,在没有显式调用 Stop 方法的情况下也可以正确地终止对象。如果调用了 Stop 方法,将不需要析构函数。SuppressFinalize 方法能够防止调用对象的 Finalize 方法(析构函数的实际实现)。

CWorkerThread 抽象类

  CWorkerThread 是一个由 CWorkerThreadAppSpecifc、CWorkerThreadRoundRobin 和 CWorkerThreadAssembly 继承的抽象类。无论如何处理消息,队列的大部分处理是相同的,所以 CWorkerThread 类提供了这一功能。这个类提供了抽象方法(必须被实际方法替代)以管理资源和处理消息。

  类的工作再一次通过 Start、Stop、Pause 和 Continue 方法来实现。在 Start 方法中引用了输入和错误队列。在 .NET 框架中,消息由 System.Messaging 名称空间处理:

// 尝试打开队列,并设置默认的读写属性
MessageQueue mqInput = new MessageQueue(sInputQueue);
mqInput.MessageReadPropertyFilter.Body = true;
mqInput.MessageReadPropertyFilter.AppSpecific = true;
MessageQueue mqError = new MessageQueue(sErrorQueue);
// 如果使用 MSMQ COM,则将格式化程序设置为 ActiveX
mqInput.Formatter = new ActiveXMessageFormatter();
mqError.Formatter = new ActiveXMessageFormatter();

  一旦定义了消息队列引用,即会创建一个线程用于实际的处理函数(称为 ProcessMessages)。在 .NET 框架中,使用 System.Threading 名称空间很容易实现线程处理:

procMessage = new Thread(new ThreadStart(ProcessMessages));
procMessage.Start();

  ProcessMessages 函数是基于 Boolean 值的处理循环。当数值设为 False,处理循环将终止。因此,线程对象的 Stop 方法只设置这一 Boolean 值,然后关闭打开的消息队列,并加入带有主线程的线程:

// 加入服务线程和处理线程
bRun = false;
procMessage.Join();
// 关闭打开的消息队列
mqInput.Close();
mqError.Close();

Pause 方法只设置一个 Boolean 值,使处理线程休眠半秒钟:

if (bPause)
Thread.Sleep(500);

  最后,每一个 Start、Stop、Pause 和 Continue 方法将调用抽象的 OnStart、OnStop、OnPause 和 OnContinue 方法。这些抽象方法为实现的类提供了挂钩,以捕获和释放所需的资源。

ProcessMessages 循环具有如下基本结构:

接收 Message。

如果 Message 具有成功的 Receive,则调用抽象 ProcessMessage 方法。

如果 Receive 或 ProcessMessage 失败,将 Message 发送至错误队列中。
Message mInput;
try
{
// 从队列中读取,并等候 1 秒
mInput = mqInput.Receive(new TimeSpan(0,0,0,1));
}
catch (MessageQueueException mqe)
{
// 将消息设置为 null
mInput = null;
// 查看错误代码,了解是否超时
if (mqe.ErrorCode != (-1072824293) ) //0xC00E001B
{
// 如果未超时,发出一个错误并记录错误号
LogError("Error: " + mqe.Message);
throw mqe;
}
}
if (mInput != null)
{
// 得到一个要处理的消息,调用处理消息抽象方法
try
{
ProcessMessage(mInput);
}
// 捕获已知异常状态的错误
catch (CWorkerThreadException ex)
{
ProcessError(mInput, ex.Terminate);
}
// 捕获未知异常,并调用 Terminate
catch
{
ProcessError(mInput, true);
}
}

  ProcessError 方法将错误的消息发送至错误队列。另外,它也可能引发异常来终止线程。如果ProcessMessage 方法引发了终止错误或 CWorkerThreadException 类型,它将执行此操作。

0
相关文章