微软消息件MSMQ研究DEMO(二)

本DEMO是一个异构系统,打算利用VC++编写一个底层处理请求的逻辑,简称为AppServer;利用VS.NET2003 编写一个Webservice中间服务;暂时命名为 MqWebService; 最后会编写一个WEB程序进行交易请求和显示.
    首先介绍下 WebService的编写:
    WebService是.NET平台下一种主流的服务提供方式,它通过以后SOAP协议传输XML格式的数据,具体的定义记不太清楚了.
    
    DEMO中MqWebService的设计思路:
    本想画个设计图的,可惜Visio没有装好,就在这里用文字叙述下吧;
      一, MyQueue类: 封装了对MSMQ的访问函数(DEMO中使用的都是同步方式)

/********************************************************************************************
 * Author:陈子文
 * CreateDate;2007-11-20
 * function:这是一个MSMQ类，定义了MSMQ queue的创建，消息的接收和发送
 * summary:这是一个同步MSMQ
 * Email:  skyczw2005@yahoo.com.cn
 * *****************************************************************************************/

using System;
using System.Configuration;
using System.Messaging;


namespace MqWebService
{
 /// <summary>
 /// MyQueue 的摘要说明。
 /// </summary>
 public class MyQueue
 {

  //成员变量
  protected MessageQueue _queue;
  protected MessageQueueTransactionType _mqTransaction;
  protected TimeSpan _timeSpan;

  //构造函数
  public MyQueue(string mqPath,int timeSpan)
  {
   _mqTransaction = MessageQueueTransactionType.Automatic;
   _timeSpan = TimeSpan.FromMilliseconds((double)timeSpan);

   //创建QUEQUE
   _queue = new MessageQueue(mqPath,true);

   //去掉不用的属性
   _queue.DefaultPropertiesToSend.AttachSenderId = false;
   _queue.DefaultPropertiesToSend.UseAuthentication = false;
   _queue.DefaultPropertiesToSend.UseEncryption = false;
   _queue.DefaultPropertiesToSend.AcknowledgeType = AcknowledgeTypes.None;
   _queue.DefaultPropertiesToSend.UseJournalQueue = false;

  }
  
  /// <summary>
  /// 从消息队列接收消息
  /// </summary>
  /// <returns></returns>
  public virtual object ReceiveMessage()
  {
   try
   {
    using( Message msg = _queue.Receive(_timeSpan,_mqTransaction))
      return msg;
   }
   catch(MessageQueueException mqEx)
   {
    if (mqEx.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
     throw new TimeoutException();
    throw;
   }
  }

  /// <summary>
  /// 发送消息到队列
  /// </summary>
  /// <param name="msg"></param>
  /// <returns></returns>
 
  public virtual  void  SendMessage(string msg)
  {
   _queue.Formatter = new ActiveXMessageFormatter();
   try
   {
    _queue.Send(msg,_mqTransaction);
   }
   catch(MessageQueueException mqEx)
   {
    throw mqEx;
   }
   catch(System.Exception ex)
   {
    throw ex;
   }
  }
 }
}



  
    二,建立好访问MQ的类后,在定义我们请求消息和回复消息的实体类:
     
/********************************************************************************************
 * Author:陈子文
 * CreateDate;2007-11-20
 * function:业务对象实体类
 * summary:
 * Email:  skyczw2005@yahoo.com.cn
 * *****************************************************************************************/

using System;
using System.Xml;
using System.Xml.Serialization;

namespace MqWebService
{
 /// <summary>
 /// InfoFormate 的摘要说明。
 /// </summary>
 [Serializable]
 public class InfoFormate
 {
  //成员变量
  private string _id;
  private string _bankNo;
  private int _money;
  private int _goodsId;


  public InfoFormate()
  {
   _id = "";
   _bankNo = "";
   _money = 0;
   _goodsId = 0 ;
  }
  
  #region 属性
  public string Id
  {
   set { _id = value;}
   get {return _id ;}
  }

  public string BankNo
  {
   set {_bankNo = value;}
   get {return _bankNo ;}
  }

  public int Money
  {
   set {_money = value;}
   get { return _money;}
  }

  public int GoodId
  {
   set { _goodsId = value;}
   get {return _goodsId;}
  }
  #endregion

  //方法
  /// <summary>
  /// 打包
  /// </summary>
  /// <param name="info"></param>
  /// <returns></returns>
  public static string InPackage(InfoFormate info)
  {
   string  _package = info.Id.Trim()+"#"+info.BankNo.Trim()+"#"+info.Money.ToString().Trim()+"#"+info.GoodId.ToString().Trim();
   return  _package;

  }
  
  //解包
  /// <summary>
  /// 
  /// </summary>
  /// <param name="package"></param>
  /// <returns></returns>
  public static InfoFormate UnPackage(string package)
  {
   InfoFormate info = new InfoFormate();
    try
    {
     string[] tmp=package.Split("#".ToCharArray(),10);
     info.Id =tmp[0].Trim();
     info.BankNo = tmp[1].Trim();
     info.Money = int.Parse(tmp[2].Trim());
     info.GoodId = int.Parse(tmp[3].Trim());
    }
    catch(System.Exception ex)
    {
     throw ex ;
    }

   return info;
  }
 }

 

 /// <summary>
 /// InfoFormate 的摘要说明。
 /// </summary>
 [Serializable]
 public class InfoRespone
 {
  //成员变量
  private string _Transid;
  private string _bankNo;
  private int _accountMoney; //帐户余额
  private int _money;
  private int _goodsId;
  private int _state;   //0 失败 1成功


  public InfoRespone()
  {
   _Transid = "";
   _bankNo = "";
   _money = 0;
   _goodsId = 0 ;
   _state = 0;
   _accountMoney = 0;

  }
  
  #region 属性
  public string TransId
  {
   set { _Transid = value;}
   get {return _Transid ;}
  }

  public string BankNo
  {
   set {_bankNo = value;}
   get {return _bankNo ;}
  }

  public int Money
  {
   set {_money = value;}
   get { return _money;}
  }

  public int GoodId
  {
   set { _goodsId = value;}
   get {return _goodsId;}
  }

  public int State
  {
   set {_state = value;}
   get {return _state;}
  }

  public int AccountMoney
  {
   set {_accountMoney = value;}
   get {return _accountMoney ;}
  }

  #endregion

  //方法
  /// <summary>
  /// 打包
  /// </summary>
  /// <param name="info"></param>
  /// <returns></returns>
  public static string InPackage(InfoRespone info)
  {
   string  _package = info.TransId.Trim()+"#"+info.BankNo.Trim()+"#"+info.Money.ToString().Trim()+"#"+info.GoodId.ToString().Trim()+"#"+info.AccountMoney+"#"+info.State;
   return  _package;

  }
  
  //解包
  /// <summary>
  /// 
  /// </summary>
  /// <param name="package"></param>
  /// <returns></returns>
  public static InfoRespone UnPackage(string package)
  {
   InfoRespone info = new InfoRespone();
   try
   {
    string[] tmp=package.Split("#".ToCharArray(),10);
    info.TransId =tmp[0];
    info.BankNo = tmp[1];
    info.Money = int.Parse(tmp[2].Trim());
    info.GoodId = int.Parse(tmp[3].Trim());
    info.AccountMoney =100; //int.Parse(tmp[4].Trim());
    info.State =1; //int.Parse(tmp[5].Trim());

   }
   catch(System.Exception ex)
   {
    throw ex ;
   }

   return info;
  }
 }
}



    这两个类是简单的对业务进行了处理,没有很详细的考虑异常等处理.

    三,建立好访问MQ的类后,在定义我们请求消息和回复消息的MSMQ类:
     
/********************************************************************************************
 * Author:陈子文
 * CreateDate;2007-11-20
 * function:这是一个业务请求MQ类，定义了消息的接收和发送
 * summary:这是一个同步MSMQ
 * Email:  skyczw2005@yahoo.com.cn
 * *****************************************************************************************/

using System;
using System.Configuration;
using System.Messaging;

namespace MqWebService
{
 /// <summary>
 /// RequstMq 的摘要说明。
 /// </summary>
 public class RequstMq :MyQueue
 {
  private static string _mqPath = "chen\\private$\\mytestmq";  //以后加
  private  static int  _timeSpan = 200 ;

  public RequstMq():base(_mqPath,_timeSpan)   //
  {
   _queue.Formatter = new ActiveXMessageFormatter();
  }
  
  /// <summary>
  /// 从消息队列接收消息
  /// </summary>
  /// <returns></returns>
  public  new  string ReceiveMessage()
  {
    return (string)((Message)base.ReceiveMessage()).Body;
  }

  /// <summary>
  /// 发送消息到队列
  /// </summary>
  /// <param name="msg"></param>
  /// <returns></returns>
 
  public   void  SendMessage(string msg)
  {
   base._mqTransaction = MessageQueueTransactionType.Single;
   base.SendMessage(msg);
  }

  /// <summary>
  /// 从消息队列接收消息
  /// </summary>
  /// <returns></returns>
  public  string ReceiveMessage(int timeSpan)
  {
   base._timeSpan = TimeSpan.FromMilliseconds((double)timeSpan);
   return ReceiveMessage();
  }

 
 }
}


///

/********************************************************************************************
 * Author:陈子文
 * CreateDate;2007-11-20
 * function:这是一个业务回复MQ类，定义了消息的接收和发送
 * summary:这是一个同步MSMQ
 * Email:  skyczw2005@yahoo.com.cn
 * *****************************************************************************************/

using System;
using System.Configuration;
using System.Messaging;

namespace MqWebService
{
 /// <summary>
 /// RequstMq 的摘要说明。
 /// </summary>
 public class ResponeMSMQ :MyQueue
 {
  private static string _mqPath = @"chen\private$\MyReceMq";  //以后加
  private static int  _timeSpan = 200;

  public ResponeMSMQ():base(_mqPath,_timeSpan)
  {
   _queue.Formatter = new ActiveXMessageFormatter();
  }
  
  /// <summary>
  /// 从消息队列接收消息
  /// </summary>
  /// <returns></returns>
  public  new  string ReceiveMessage()
  {
   base._mqTransaction = MessageQueueTransactionType.Automatic;
   string tmp ="";
   try
   {
    tmp = ((Message)base.ReceiveMessage()).Body.ToString();
   }
   catch(System.Exception ex)
   {
    throw ex;
   }

   return tmp;
  }

  /// <summary>
  /// 发送消息到队列
  /// </summary>
  /// <param name="msg"></param>
  /// <returns></returns>
 
  public   void  SendMessage(string msg)
  {
   base._mqTransaction = MessageQueueTransactionType.Single;
   base.SendMessage(msg);
  }

  /// <summary>
  /// 从消息队列接收消息
  /// </summary>
  /// <returns></returns>
  public  string ReceiveMessage(int timeSpan)
  {
   base._timeSpan = TimeSpan.FromMilliseconds((double)timeSpan);
   return  ReceiveMessage();
  }

 
 }
}



    四,现在可以定义一个任务类Task: 这个类是每个请求上来的实体,里面封装了,对消息的处理和请求的状态等等;
     
using System;
using System.Xml;
using System.Xml.Serialization;

namespace MqWebService
{
 /// <summary>
 /// Task 的摘要说明。
 /// </summary>
 public class Task
 {

  //成员变量
  private string _Idle;
  private InfoFormate _infoFormate;
  private InfoRespone _infoRespone;
  private TaskState _state;
  private BLL _bll;

  public Task(BLL bll)
  {
   _Idle ="";
   _state = TaskState.idle;
   this._bll = bll;
   _infoRespone = new InfoRespone();
  }

  public string Idle
  {
   get{return Idle;}
  }

  public InfoRespone Request(InfoFormate info)
  {
   lock(this)
   {
    _Idle = info.Id;
    _infoFormate = info;
    _state = TaskState.run;

    //初试回复
    _infoRespone.TransId = _infoFormate.Id;
    _infoRespone.BankNo = _infoFormate.BankNo;
    _infoRespone.Money = _infoFormate.Money;
    _infoRespone.State = 0 ;
   }

   //打包
   string _requestStr = InfoFormate.InPackage(_infoFormate);
   
   //处理中心处理
   if ( _bll.Request(this,_requestStr) == true )
   {
    WaitForProcess(20);
   }
   else
   {
    _infoRespone.State = 0 ;
   }

   return _infoRespone;

  }

  //等待系统处理
  public void WaitForProcess(int outTime)
  {
   int time = 0;
   while(true)
   {
    
    if (_state == TaskState.finnish )
    {
     _infoRespone.State = 1;
     break;
    }

    if (time > outTime)
    {
     _infoRespone.State = 0;
     _bll.Unregister(this);
     break;
    }
    
    time ++ ;

    System.Threading.Thread.Sleep(100);
   }
   
  }

  //收到回复
  public  void OnRespone( InfoRespone info)
  {
   lock(this)
   {
    this._infoRespone = info;
    _state = TaskState.finnish;
   }
  }
 }

 public enum TaskState
 {
  idle,
  run,
  finnish
 }
}



    五,最后定义我们的处理中心了BLL,BLL里面含有一个线程和一个ArrayList;ArrayList是我们的请求处理队列;线程是持续的读回复MSMQ消息;
     
using System;
using System.Collections;
using System.Threading;
using System.Xml;
using System.Xml.Serialization;

namespace MqWebService
{
 /// <summary>
 /// BLL 的摘要说明。
 /// </summary>
 public class BLL
 {

  //成员变量
  private static ArrayList _table = new ArrayList();
  private Thread _thread;
  private RequstMq _requestMq;
  private ResponeMSMQ _responeMSMQ;
  private bool _continueFlag;

  public BLL()
  {
   _continueFlag = false;
   _thread = new Thread(new ThreadStart(Working));
   _requestMq = new RequstMq();
   _responeMSMQ = new ResponeMSMQ();
    RunService();

  }


  public void RunService()
  {
   _continueFlag = true;
   _thread.Start();
  }

  public void StopService()
  {
   _continueFlag = false;
   _thread.Suspend();
  }

  public void Unregister(Task task)
  {
   lock(this)
   {
    _table.Remove(task);
   }
  }

  public void Working()
  {
    while(_continueFlag)
    {
     try
     {
   
     string msg = _responeMSMQ.ReceiveMessage();
     if (msg.Trim().Length == 0)
      continue;

     InfoRespone info = InfoRespone.UnPackage(msg);

     Task _task = null;
     if (info !=null)
     {
      lock(this)
      {
       for(int i=0; i<_table.Count ; i++)
       {
        Task tmp = (Task)_table[i];
        if (tmp.Idle == info.TransId )
        {
         _task = tmp;
         _table.Remove(tmp);
         break;
        }
       }

      }
     }

     if (_task != null)
     {
      _task.OnRespone(info);
     }

     Thread.Sleep(100);
    }
   catch(System.Exception ex)
   {
    //throw ex;
   }
  }
 

   
  }

  public bool Request(Task task, string msg)
  {
   bool retcode= false;

   try
   {
    _requestMq.SendMessage(msg);
    _table.Add(task);
     retcode= true;

   }
   catch(System.Exception ex)
   {
    throw ex;
   }

   return retcode;

  }
 }
}



    ok,WebService端的基础代码完成了.