消息队列（MSMQ）实现多服务器应用程序之间消息实时交互
我所介绍的例子是利用微软的消息队列（msmq）实现多个服务器之间消息实时传递。
应用程序：基于dotnet平台采用WinForm+Webservice开发的应用程序。
每个地区都有自己的数据库和Webservice服务器。
Webservice服务器有很多台，这样多个服务器上用户互相交流就成了问题。思前想后采用了msmq，设计思想如下：
例子：A服务器用户user1发送消息给B服务器上的user2
1、首先是数据库结构是一样的使用sql2005同步用户信息表。
2、在每台Webservice服务器上安装msmq。
3、当user1发送消息给user2时，判断user2所在服务器，如果user2和发送者不在同一个服务器就使用msmq传递消息给B服务器。
4、每个Webservice服务器都有一个消息接收服务程序，用来侦测本服务器消息队列里的消息，反序列化消息内容，写入数据库。
5、user2直接读取数据库就可以了。
6、实际上就是通过msmq在多个数据库服务器之间消息传递。
下面是消息接收服务程序的源代码：
Codeusing System;
using System.Collections;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.ServiceProcess;
using System.Threading;
using System.Messaging;
using wzh.Db;//数据库操作类
///<summary>
///名称：消息服务程序
///用途：实时侦测远程机器发送到本机的消息队列的消息，写入本机数据库。
///作者：wzh
///编写日期：2008-9-28
///修改日期：wzh于2008-10-10修改了部分BUG
///</summary>
namespace wzh_MsgService
{
public class MsgService : System.ServiceProcess.ServiceBase
{
/// <summary>
/// 必需的设计器变量。
/// </summary>
private System.ComponentModel.Container components = null;
private bool servicePaused;
private string mPath = null;// = @".\Private$\MsgQueue
public MsgService()
{
// 该调用是 Windows.Forms 组件设计器所必需的。
InitializeComponent();
// TODO: 在 InitComponent 调用后添加任何初始化
}
// 进程的主入口点
static void Main()
{
System.ServiceProcess.ServiceBase[] ServicesToRun;
// 同一进程中可以运行多个用户服务。若要将
//另一个服务添加到此进程，请更改下行
// 以创建另一个服务对象。例如，
//
// ServicesToRun = New System.ServiceProcess.ServiceBase[] {new Service1(), new MySecondUserService()};
//
ServicesToRun = new System.ServiceProcess.ServiceBase[] { new MsgService() };
System.ServiceProcess.ServiceBase.Run(ServicesToRun);
}
/// <summary>
/// 调试用在项目属性里，将输出类型(Output Type)改成Console Application，将Startup Object改成Namespace.ServiceName
/// </summary>
/// <param name="args"></param>
// public static void Main(string[] args)
// {
// MsgService s = new MsgService();
// s.OnStart(args);
// Console.ReadLine();
// s.OnStop();
// }
/// <summary>
/// 设计器支持所需的方法 - 不要使用代码编辑器
/// 修改此方法的内容。
/// </summary>
private void InitializeComponent()
{
components = new System.ComponentModel.Container();
this.ServiceName = "QZCPNET_MsgService";
this.CanPauseAndContinue = true;
this.CanStop = true;
}
/// <summary>
/// 清理所有正在使用的资源。
/// </summary>
protected override void Dispose( bool disposing )
{
if( disposing )
{
if (components != null)
{
components.Dispose();
}
}
base.Dispose( disposing );
}
/// <summary>
/// 设置具体的操作，以便服务可以执行它的工作。
/// </summary>
protected override void OnStart(string[] args)
{
// TODO: 在此处添加代码以启动服务。
servicePaused=false;
mPath = GetXmlData("//msgmsmqpath");
Thread newThread=new Thread(new ThreadStart(RecevieMessage));
newThread.IsBackground=true;
newThread.Start();
}
/// <summary>
/// 接收本地服务器上的MSMQ里的消息
/// </summary>
private void RecevieMessage()
{
if(!servicePaused)
{
if(! MessageQueue.Exists(mPath))
{
MessageQueue.Create(mPath);
MessageQueue mqTemp = new MessageQueue(mPath);
mqTemp.SetPermissions("Everyone", MessageQueueAccessRights.FullControl);
}
MessageQueue MyQueue = new MessageQueue(mPath);
MyQueue .Formatter =new XmlMessageFormatter (new Type []{typeof (DataSet)});
MyQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(MyReceiveCompleted);
MyQueue.BeginReceive();
}
}
private void MyReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
{
MessageQueue mq = (MessageQueue)source;
try
{
Message ms = mq.EndReceive(asyncResult.AsyncResult);
DataSet ds = (DataSet)ms.Body;
ProcessMsg(ds);
}
catch(Exception ex)
{
if(ex.Source != "System.Messaging")
WriteSysLog(ex.ToString());
}
finally
{
GC.Collect();
mq.BeginReceive();
}
return;
}
/// <summary>
/// 消息处理
/// </summary>
/// <param name="ds"></param>
private void ProcessMsg(DataSet ds)
{
//插入新消息
try
{
clsDBExec clsDb=new clsDBExec(GetXmlData("//conn") ,"消息管理-SaveMsg");
string strSql="";
DataRow dr=ds.Tables[0].Rows[0];
strSql="INSERT INTO SM_MsgessSend ( MsgSId, MsgTitle, MsgSender,MsgSenderJgCode, MsgSenderJg, MsgAderrs,MsgSAderrs ,MsgContent, "+
" MsgAccessName, MsgAccessDes, MsgHz)"+
" VALUES ("+dr["MsgSId"].ToString()+",\'"+ dr["MsgTitle"].ToString()+"\',\'"+ dr["MsgSender"].ToString() +"\',\'"+ dr["MsgSenderJgCode"].ToString() + "\',\'" + dr["MsgSenderJg"].ToString() +"\',\'"+ dr["MsgAderrs"].ToString() +"\',\'"+
dr["MsgSAderrs"].ToString() +"\',\'"+dr["MsgContent"].ToString().Replace("\'","\'\'") +"\',\'"+ dr["MsgAccessName"].ToString().Replace("\'","\'\'") +"\',\'"+ dr["MsgAccessDes"].ToString() +"\',\'"+dr["MsgHz"].ToString() + "\');";
strSql+="SELECT @@IDENTITY ";
string strV=clsDb.getOneVal(strSql);
//发入发件箱
strSql="INSERT INTO SM_MsgessUser(UserId, MsgId,MsgFolder,MsgIsRead ) values(\'"+ dr["MsgSender"].ToString() +"\',"+ strV +",\'发件箱\',2)";
clsDb.execNonQuery(strSql);
//插入到用户
string strAderrs=dr["MsgSAderrs"].ToString();
string [] arrStr=strAderrs.Split(\',\');
strSql="";
foreach(string strUser in arrStr)
{
if (strUser.ToUpper()=="ALL" )
{
strSql="INSERT INTO SM_MsgessUser(UserId, MsgId,MsgFolder ) SELECT UserName,"+ strV +",\'收件箱\' FROM SM_UserId";
break;
}
if (strUser.ToUpper()=="ALLONLINE" )
{
strSql="INSERT INTO SM_MsgessUser(UserId, MsgId,MsgFolder ) SELECT DISTINCT UserId,"+ strV +",\'收件箱\' FROM SM_OnlineUser";
break;
}
if ((strUser.ToUpper()!="ALL" ) && strUser.ToUpper()!="ALLONLINE")
{
strSql+="INSERT INTO SM_MsgessUser(UserId, MsgId ,MsgFolder) SELECT UserName,"+ strV +",\'收件箱\' FROM SM_UserId where UserName=\'"+ strUser +"\';";
}
}
clsDb.execNonQuery(strSql);
}
catch(Exception ex)
{
WriteSysLog(ex.Message);
}
}
/// <summary>
/// 停止此服务。
/// </summary>
protected override void OnStop()
{
// TODO: 在此处添加代码以执行停止服务所需的关闭操作。
servicePaused=true;
}
/// <summary>
/// 获取XML文件内容
/// </summary>
/// <param name="nodepath"></param>
/// <returns></returns>
private string GetXmlData(string nodepath)
{
System.Xml.XmlDocument dom=new System.Xml.XmlDocument();
dom.Load(@"e:\net\service\Serverconfig.xml");
return dom.SelectSingleNode(nodepath).InnerText;
}
/// <summary>
/// 写日志
/// </summary>
/// <param name="strMsg"></param>
private void WriteSysLog(string strMsg)
{
EventLog objLog=new EventLog("Application") ;
objLog.Source ="信息网络管理系统--消息服务";
objLog.WriteEntry(strMsg,EventLogEntryType.Error );
}
}
}