.Net下的MSMQ的同步异步调用_.Net教程

以下为引用的内容：
1using System;
2using System.Messaging;
3using System.Threading;

5
6namespace LoveStatusService
7{
8 /**//// <summary>
9 /// Summary description for Msmq.
10 /// </summary>
11 public class Msmq
12 {
13 public Msmq()
14 {
15 //
16 // TODO: Add constructor logic here
17 //
18 }
19
20
21 private MessageQueue _messageQueue=null;
22 //最大并发线程数
23 private static int MAX_WORKER_THREADS=Convert.ToInt32( System.Configuration.ConfigurationSettings.AppSettings["MAX_WORKER_THREADS"].ToString());
24 //Msmq路径
25 private static string MsmqPath=System.Configuration.ConfigurationSettings.AppSettings["LoveStatusMQPath"];
26 //等待句柄
27 private WaitHandle[] waitHandleArray = new WaitHandle[MAX_WORKER_THREADS];
28 //任务类型
29 //1. Send Email 2. Send Message 3. Send Email and Message
30 private string TaskType=System.Configuration.ConfigurationSettings.AppSettings["TaskType"];
31 public MessageQueue MessQueue
32 {
33 get
34 {
35
36 if (_messageQueue==null)
37 {
38 if(MessageQueue.Exists(MsmqPath))
39 {
40 _messageQueue = new MessageQueue(MsmqPath);
41 }
42 else
43 {
44 _messageQueue = MessageQueue.Create(MsmqPath);
45 }
46 }
47
48
49 return _messageQueue;
50 }
51 }
52
53
54 Private Method#region Private Method
55
56 private void mq_ReceiveCompleted(object sender, System.Messaging.ReceiveCompletedEventArgs e)
57 {
58 MessageQueue mqq = (MessageQueue)sender;
59 System.Messaging.Message m = mqq.EndReceive(e.AsyncResult);
60 //m.Formatter = new System.Messaging.XmlMessageFormatter(new string[]{"System.String,mscorlib"});
61 m.Formatter =new System.Messaging.XmlMessageFormatter(new Type[] {typeof(UserObject)}) ;
62 //log.Info("Receive UserID: " (string)m.Body) ;
63 UserObject obj=(UserObject)m.Body ;
64 long curUserId=obj.curUserID ;
65 long oppUserId=obj.oppUserID;
66 string curUserName=obj.curUserName;
67 string oppUserName=obj.oppUserName;
68 string curEmail=obj.curEmail ;
69 string oppEmail=obj.oppEmail;
70 string subject =obj.subject ;
71 string body=obj.body ;
72 //AppLog.log.Info("curUserId:" curUserId) ;
73 //AppLog.log.Info("oppUserId:" oppUserId) ;
74 AppLog.log.Info("==type=" TaskType) ;
75 switch(TaskType)
76 {
77 //Email
78 case "1":
79 EmailForMQ.SendEmailForLoveStatus(curEmail,oppEmail,curUserName,oppUserName,subject) ;
80 AppLog.log.Info("==Send to==" oppEmail) ;
81 break;
82 //Message
83 case "2":
84 MessageForMQ.SendMessage(curUserId,oppUserId,subject,body) ;
85 AppLog.log.Info("==Send Msg to==" oppUserId) ;
86 break;
87 //Email and Message
88 case "3":
89 EmailForMQ.SendEmailForLoveStatus(curEmail,oppEmail,curUserName,oppUserName,subject) ;
90 AppLog.log.Info("==Send to==" oppEmail) ;
91 MessageForMQ.SendMessage(curUserId,oppUserId,subject,body) ;
92 AppLog.log.Info("==Send Msg to==" oppUserId) ;
93 break;
94 default:
95 break;
96
97 }
98 mqq.BeginReceive() ;
99
100 }
101
102 #endregion
103
104 Public Method#region Public Method
105
106 //一个将对象发送到队列的方法,这里发送的是对象
107 public void SendUserIDToMQ(object arr)
108 {
109 MessQueue.Send(arr) ;
110 Console.WriteLine("Ok") ;
111 Console.Read() ;
112 }
113
114 //同步接受队列内容的方法
115 public void ReceiveFromMQ()
116 {
117 Message ms=new Message() ;
118
119 //ms=MessQueue.Peek();
120 try
121 {
122 ms=MessQueue.Receive(new TimeSpan(0,0,5));
123 if(ms!=null)
124 {
125 ms.Formatter = new XmlMessageFormatter ( new Type [] { typeof (string) } );
126 AppLog.log.Info((string)ms.Body) ;
127 }
128 }
129 catch(Exception ex)
130 {
131
132 }
133
134
135 }
136
137 //开始监听工作线程
138 public void startListen()
139 {
140 AppLog.log.Info("--Thread--" MAX_WORKER_THREADS) ;
141 MessQueue.ReceiveCompleted =new ReceiveCompletedEventHandler(mq_ReceiveCompleted);
142
143 //异步方式，并发
144
145 for(int i=0; i<MAX_WORKER_THREADS; i )
146 {
147 // Begin asynchronous operations.
148 waitHandleArray[i] = MessQueue.BeginReceive().AsyncWaitHandle;
149 }
150
151 AppLog.log.Info("------Start Listen--------") ;
152
153 return;
154
155 }
156
157
158 //停止监听工作线程
159 public void stopListen()
160 {
161
162 for(int i=0;i<waitHandleArray.Length;i )
163 {
164
165 try
166 {
167 waitHandleArray[i].Close();
168 }
169 catch
170 {
171 AppLog.log.Info("---waitHandleArray[i].Close() Error!-----") ;
172 }
173
174 }
175
176 try
177 {
178 // Specify to wait for all operations to return.
179 WaitHandle.WaitAll(waitHandleArray,1000,false);
180 }
181 catch
182 {
183 AppLog.log.Info("---WaitHandle.WaitAll Error!-----") ;
184 }
185 AppLog.log.Info("------Stop Listen--------") ;
186
187 }
188
189 #endregion
190
191
192
193
194 }
195}
196