手写MSMQ微软消息队列收发工具类

骑只蚂蚁游世界

一、MSMQ介绍
MSMQ（Microsoft Message Queuing）是微软开发的消息队列技术，支持事务，支持异步发送和接收消息。
两个重要的概念：队列和消息。队列是存放消息的容器和传输消息的通道。消息是在队列上存储和传输的数据的基本单元；这个消息在计算机上的存在形式可以是任意格式的文件；在C#程序中的消息是各种类型的Class类的实例，在程序中的消息类型可以在创建XmlMessageFormatter实例时指定。消息最大4M。
应用场景可以是异步处理，如系统短时间收到大量数据/请求，到达程序能够处理的请求数上限，可以将待处理的数据/请求全部放入队列中，程序再从队列中读取消息逐个处理。应用场景也可以是系统解耦，最常见的举例如电商平台中，订单系统要将订单数据发给支付系统时，订单系统可以将数据存入队列中，支付系统从队列中读取后处理。这时订单系统不需要调用支付系统的接口，这样订单系统和支付系统可以独立部署减少了依赖。
二、工具类封装
封装的好处：甲方早期项目中大量用到微软的技术，消息队列都是用的MSMQ。我们的多个项目都有对接MQ的需求，将消息收发的功能从多个项目中提取出来放一个独立程序中，后期维护中只需要修改一份代码。同时MQ这部分功能也从业务系统中剥离出来，更新MQ程序时不影响业务系统。封装中需要兼容的两点：

• 功能较全，适用性强。满足多种格式的数据收发，同时兼容事务性队列和非事务性队列。
  
• 配置简单。
  
• 同时支持多个队列的收发处理。
   
  

几个核心类

• MQHandler：抽象类，实现消息发送和备份、消息接收的主流程，抽象方法Send和Receive，这两个方法在下一层的实现类中实现具体功能。为什么要有这个抽象类？这里主要考虑项目中将来可能出现别的类型队列，但对【发送-备份-接收】的主流程来说不管什么类型的队列都不变，那么这部分功能对不同队列来说是相同的代码，因此在MQHandler实现，因队列类型不同的部分代码分别在下一层中实现。
  

1.    public abstract class MQHandler
2.     {
3.         protected MqCfgEntity cfg = null;
4.         //是否正在处理消息
5.         private bool IsProcessingMessage = false;
7.         public MQHandler(MqCfgEntity cfg)
8.         {
9.             this.cfg = cfg;
10.         }
12.         public void Start()
13.         {
14.             while (true)
15.             {
16.                 try
17.                 {
18.                     if (IsProcessingMessage) return;
19.                     IsProcessingMessage = true;//正在处理消息
20.                     if ("Send".Equals(cfg.ProcessType))
21.                     {
22.                         Send();
23.                     }
24.                     else if ("Receive".Equals(cfg.ProcessType))
25.                     {
26.                         Receive();
27.                     }
28.                     IsProcessingMessage = false; //消息处理完成
29.                 }
30.                 catch (Exception ex)
31.                 {
32.                     Log4Net.Error($"{ex.Message},{ex.StackTrace}");
33.                     IsProcessingMessage = false; //消息处理完成
34.                 }
35.                 Thread.Sleep(cfg.SplitSeconds * 1000);
36.             }
37.         }
39.         /// <summary>
40.         /// 备份已发送的文件,file为待备份完整文件名
41.         /// </summary>
42.         protected void BackSendFile(string file)
43.         {
44.             if (File.Exists(file) && cfg.BackPath.IsNotNullOrEmpty())
45.             {
46.                 FileInfo info = new FileInfo(file);
47.                 string backPath = cfg.BackPath;
48.                 if (cfg.BackFormat.IsNotNullOrEmpty())
49.                 {
50.                     backPath = $"{backPath}/{DateTime.Now.ToString(cfg.BackFormat)}";
51.                     if (!Directory.Exists(backPath))
52.                     {
53.                         Directory.CreateDirectory(backPath);
54.                     }
55.                 }
56.                 string backName = $"{backPath}/{info.Name}";
57.                 if (File.Exists(backName))
58.                 {
59.                     backName = $"{backPath}/{Guid.NewGuid().ToString()}_{info.Name}";
60.                 }
61.                 File.Move(file, backName);
62.             }
63.         }
65.         private void Send()
66.         {
67.             var names = Directory.GetFiles(cfg.FilePath, cfg.FileFilter);
68.             if (names != null && names.Count() > 0)
69.             {
70.                 var files = names.ToList().Select(f => f.ForMatPath());
71.                 foreach (var file in files)
72.                 {
73.                     if (file.TryOpenFile() && Send(file))
74.                     {
75.                         Log4Net.Info($"{file}已发送");
76.                         BackSendFile(file);
77.                         Log4Net.Info($"{file}已备份");
78.                     }
79.                 }
80.             }
81.         }
83.         public abstract bool Send(string file);
85.         public abstract void Receive();
86.     }

复制代码

View Code

• MSMQHandler，继承自MQHandler，实现Send和Receive方法。这里面的XmlMessageFormatter参数用来指定消息类型。如文件byte[]，文本string，xml对象XmlDocument。
  

1.    public class MSMQHandler : MQHandler
2.     {
3.         private XmlMessageFormatter formatter = null;
4.         //把xml当作txt读取在msmq中传输时，使用utf8编码，Unicode可能会造成部分报文数据紊乱
5.         private Encoding encoding = Encoding.UTF8;
7.         public MSMQHandler(MqCfgEntity cfg) : base(cfg)
8.         {
9.             Type type = GetMsgType(cfg.MessageType);
10.             formatter = new XmlMessageFormatter(new Type[] { type });
11.         }
13.         public override void Receive()
14.         {
15.             MessageQueue queue = null;
16.             try
17.             {
18.                 queue = new MessageQueue(cfg.Queue);
19.                 int num = queue.GetAllMessages().Length;
20.                 for (int i = 0; i < num; i++)
21.                 {
22.                     ReceiveMessage(queue);
23.                 }
24.             }
25.             catch (Exception ex)
26.             {
27.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
28.             }
29.             finally
30.             {
31.                 if (queue != null) queue.Dispose();
32.             }
33.         }
35.         private void ReceiveMessage(MessageQueue queue)
36.         {
37.             System.Messaging.Message message = null;
38.             try
39.             {
40.                 message = queue.Receive();
41.                 message.Formatter = formatter;
42.                 string toFile = $"{cfg.FilePath}/{message.Label}";
43.                 if ("file".Equals(cfg.MessageType))
44.                 {
45.                     SaveMessageAsBinaryFile(message, toFile);
46.                 }
47.                 else if ("xml".Equals(cfg.MessageType))
48.                 {
49.                     var doc = (XmlDocument)message.Body;
50.                     doc.Save(toFile);
51.                 }
52.                 else if ("txt".Equals(cfg.MessageType))
53.                 {
54.                     var txt = (string)message.Body;
55.                     SaveMessageAsTxt(message, toFile);
56.                 }
57.                 Log4Net.Info($"收到消息,已保存,{toFile}");
58.             }
59.             catch (Exception ex)
60.             {
61.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
62.             }
63.             finally
64.             {
65.                 if (message != null) message.Dispose();
66.             }
67.         }
69.         private void SaveMessageAsTxt(Message message, string toFile)
70.         {
71.             FileStream fs = null;
72.             try
73.             {
74.                 fs = new FileStream(toFile, FileMode.Create);
75.                 string content = (string)message.Body;
76.                 var bts = encoding.GetBytes(content);
77.                 fs.Write(bts, 0, bts.Length);
78.             }
79.             catch (Exception ex)
80.             {
81.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
82.             }
83.             finally
84.             {
85.                 if (fs != null) fs.Dispose();
86.             }
87.         }
89.         private void SaveMessageAsBinaryFile(Message message, string toFile)
90.         {
91.             FileStream fs = null;
92.             try
93.             {
94.                 fs = new FileStream(toFile, FileMode.Create);
95.                 var bts = (byte[])message.Body;
96.                 fs.Write(bts, 0, bts.Length);
97.             }
98.             catch (Exception ex)
99.             {
100.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
101.             }
102.             finally
103.             {
104.                 if (fs != null) fs.Dispose();
105.             }
106.         }
108.         public override bool Send(string file)
109.         {
110.             bool success = true;
111.             FileInfo fileInfo = new FileInfo(file);
112.             MessageQueue myQueue = null;
113.             try
114.             {
115.                 myQueue = new MessageQueue(cfg.Queue);
116.                 object body = null;
117.                 if ("file".Equals(cfg.MessageType))
118.                 {
119.                     FileStream fs = null;
120.                     try
121.                     {
122.                         fs = new FileStream(file, FileMode.Open);
123.                         byte[] bts = new byte[fs.Length];
124.                         fs.Read(bts, 0, bts.Length);
125.                         body = bts;
126.                     }
127.                     catch (Exception ex)
128.                     {
129.                         Log4Net.Error($"{ex.Message},{ex.StackTrace}");
130.                     }
131.                     finally
132.                     {
133.                         if (fs != null) fs.Dispose();
134.                     }
135.                 }
136.                 else if ("xml".Equals(cfg.MessageType))
137.                 {
138.                     XmlDocument doc = new XmlDocument();
139.                     doc.Load(file);
140.                     body = doc;
141.                 }
142.                 else if ("txt".Equals(cfg.MessageType))
143.                 {
144.                     FileStream fs = null;
145.                     try
146.                     {
147.                         fs = new FileStream(file, FileMode.Open);
148.                         byte[] bts = new byte[fs.Length];
149.                         fs.Read(bts, 0, bts.Length);
150.                         string content = encoding.GetString(bts);
151.                         body = content;
152.                     }
153.                     catch (Exception ex)
154.                     {
155.                         Log4Net.Error($"{ex.Message},{ex.StackTrace}");
156.                     }
157.                     finally
158.                     {
159.                         if (fs != null) fs.Dispose();
160.                     }
161.                 }
162.                 Push(fileInfo.Name, myQueue, body);
163.             }
164.             catch (Exception ex)
165.             {
166.                 success = false;
167.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
168.             }
169.             finally
170.             {
171.                 if (myQueue != null) myQueue.Dispose();
172.             }
173.             return success;
174.         }
176.         //往队列上推送消息
177.         private void Push(string fileName, MessageQueue myQueue, object body)
178.         {
179.             System.Messaging.Message message = null;
180.             try
181.             {
182.                 message = new System.Messaging.Message(body);
183.                 message.Formatter = formatter;
184.                 message.Label = fileName;
185.                 if (cfg.IsTransQueue)
186.                 {
187.                     using (MessageQueueTransaction trans = new MessageQueueTransaction())
188.                     {
189.                         trans.Begin();
190.                         myQueue.Send(message, trans);
191.                         trans.Commit();
192.                     }
193.                 }
194.                 else
195.                 {
196.                     myQueue.Send(message);
197.                 }
198.             }
199.             catch (Exception ex)
200.             {
201.                 Log4Net.Error($"{ex.Message},{ex.StackTrace}");
202.             }
203.             finally
204.             {
205.                 if (message != null) message.Dispose();
206.             }
207.         }
209.         /// <summary>
210.         /// 根据配置文件的类型，返回MQ队列上的消息类型
211.         /// </summary>
212.         private Type GetMsgType(string code)
213.         {
214.             Type type = null;
215.             switch (code)
216.             {
217.                 case "file": type = typeof(byte[]); break;
218.                 case "txt": type = typeof(string); break;
219.                 case "xml": type = typeof(XmlDocument); break;
220.             }
221.             return type;
222.         }
223.     }

复制代码

View Code

• XmlNodeSerializer：实体与xml互转，用于解析自定义的配置文件。代码见公众号《手写xml序列化与反序列化工具类》
  

 
配置文件说明

• mq.xml，在exe同级目录中，根节点为Config，其中可以包含多个Msmq节点，一个Msmq节点对应一个接收或发送任务。
  
• Msmq节点字段说明：
  
  
  
  • ProcessType：Send或Receive，表示用于发送或接收消息。
    
  • Queue：队列名称。
    
  • FilePath：待发送的文件所在目录，或接收到的文件的存放目录。
    
  • FileFilter：Send时才配置，表示待发送目录中哪些后缀格式的文件需要处理，如*.txt，*.xml，*.jpg，*.*。
    
  • SplitSeconds：每一轮任务处理完成后暂停多少秒再进入下一个轮循。
    
  • BackPath：Send时才配置，消息发送以后文件备份到哪个目录。
    
  • BackFormat：跟BackPath配合使用，BackPath是备份目录，BackPath表示备份文件在BackPath下按小时/天/月/年来分文件夹备份。可以为yyyyMM、yyyyMMdd等。
    
  • MessageType：消息类型，可以为file、xml、txt，表示消息以哪种类型(对应XmlMessageFormatter中的文件byte[]、文本string、xml对象XmlDocument)发送。
    
  • IsTransQueue：true或false，表示队列是否为事务性队列。
    
  
  

其它说明

• 程序运行环境：.net framework 4.5+
  
• 程序启动：直接运行MsmqClient.exe，后台进程，无前台界面。
  
• 完整项目代码：关注以下公众号，后台回复"msmq"获取
  

来源:https://www.cnblogs.com/cy2011/p/18472959
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！