请教一个关于消息队列应用的问题,谢谢!

jcs4989

首先谢谢大家的关注!

本人写了一个使用消息队列的程序大致如下:

while(1){
        if(!switcher){                                                                        (1)
                 sleep(1);                                                                      (2)
                 continue;                                                                      (3)
        }                                                                                            (4)
                                                                                                   
        int rev = msgrcv(msqid,buf,msg_len,msg_type,0);               (5)
         ...
}

说明:上述代码是在一个线程里,我的想法是如果开关switcher打开了(也就是为1)那么允许该线程接收消息队列中的消息,否则sleep一秒后再重新检测.直到开关打开.
        其中switcher开关由另外一个线程进行动态设置,在本线程中只是读取该变量值.

上面代码的问题是:
        如果其它线程在某一时刻将开关打开,并且本线程运行到(5)时没有收到指定类型的消息,那么该线程将在(5)处阻塞.
        此后某时刻另外线程又将开关关闭了.然后指定消息类型消息到达.
        那么就会出现虽然开关switcher关闭,但是仍然接收到消息的情况.这是我不想得到的.
求助:
        我想保留消息队列中如果指定类型消息没有到达,那么阻塞这一特性(不想设置IPC_NOWAIT).同时想实现我的设计目标.那么我该如何做呢?


                                                                             感谢您的指点

lipeng_927

我自己写的一个队列，放在c++版里没人看，你看看能不能用的上

1. 
   
2. 
   
3. #ifdef WIN32
   
4. #include <winbase.h>
   
5. #else
   
6. #include <pthread.h>
   
7. #include <unistd.h>
   
8. #endif
   
9. 
   
10. #include <errno.h>
    
11. #include <time.h>
    
12. #include <sys/types.h>
    
13. #include <queue>
    
14. using namespace std;
    
15. 
    
16. //这个是线程安全的队列，Synchronized类的实现在下面
    
17. template <class Type>
    
18. class SyncQueue:public Synchronized{
    
19. public:
    
20.         SyncQueue();
    
21.         ~SyncQueue();
    
22.         void push(const Type &);
    
23.         Type pop();
    
24. private:
    
25.         queue<Type> *p_Queue;
    
26. };
    
27. 
    
28. 
    
29. template <class Type>
    
30. SyncQueue<Type>::SyncQueue(){
    
31.         p_Queue = new queue<Type>() ;
    
32. }
    
33. 
    
34. template <class Type>
    
35. SyncQueue<Type>::~SyncQueue(){
    
36.         delete p_Queue;
    
37. }
    
38. 
    
39. template <class Type>
    
40. Type SyncQueue<Type>::pop(){
    
41.         Type type;
    
42.         lock();
    
43.         while(p_Queue->empty()){
    
44.                 try
    
45.                 {
    
46.                         wait();
    
47.                 }
    
48.                 catch(exception& ex)
    
49.                 {
    
50.                         cout<< ex.what()<<endl;
    
51.                         throw;
    
52.                 }               
    
53.         }
    
54.         type=p_Queue->front();
    
55.         p_Queue->pop();
    
56.         unlock();
    
57.         return type;
    
58. }
    
59. 
    
60. template <class Type>
    
61. void SyncQueue<Type>::push(const Type &type){
    
62.         lock();
    
63.         p_Queue->push(type);
    
64.         try
    
65.         {
    
66.                 notify();
    
67.         }
    
68.         catch(exception& ex)
    
69.         {
    
70.                 cout<<ex.what()<<endl;
    
71.                 throw;
    
72.         }               
    
73.         unlock();
    
74. }
    
75. 
    
76. //Synchronized类的实现，仿java的一些方法，我参考了别人的代码
    
77. 
    
78. class Synchronized
    
79. {
    
80. public:
    
81.         Synchronized();
    
82.         ~Synchronized();
    
83. 
    
84. 
    
85.         // Causes current thread to wait until another thread
    
86.         // invokes the notify() method or the notifyAll()
    
87.         // method for this object.
    
88.         void                wait();
    
89. 
    
90. 
    
91.         // Causes current thread to wait until either another
    
92.         // thread invokes the notify() method or the notifyAll()
    
93.         // method for this object, or a specified amount of time
    
94.         // has elapsed.
    
95.         // @param timeout
    
96.         //    timeout in milliseconds.
    
97.         // @param
    
98.         //    return TRUE if timeout occured, FALSE otherwise.
    
99.         bool                wait(unsigned long timeout);   
    
100.         // Wakes up a single thread that is waiting on this
     
101.         // object's monitor.
     
102.         void                notify();
     
103.         // Wakes up all threads that are waiting on this object's
     
104.         // monitor.
     
105.         void                notify_all();
     
106.         // Enter a critical section.
     
107.         void                lock();
     
108.         // Try to enter a critical section.
     
109.         // @return TRUE if the attempt was successful, FALSE otherwise.
     
110.         bool                trylock();
     
111.         // Leave a critical section.
     
112.         void                unlock();
     
113. 
     
114. private:
     
115. #ifdef WIN32
     
116.         char            numNotifies;
     
117.         HANDLE          semEvent;
     
118.         HANDLE          semMutex;
     
119.         bool                        isLocked;
     
120. #else
     
121.         int        cond_timed_wait(const timespec*);
     
122.         pthread_cond_t  cond;
     
123.         pthread_mutex_t monitor;
     
124. #endif
     
125. };
     
126. 
     
127. Synchronized::Synchronized()
     
128. {
     
129. #ifdef WIN32
     
130.         // Semaphore initially auto signaled, auto reset mode, unnamed
     
131.         semEvent = CreateEvent(0, false, false, 0);
     
132.         // Semaphore initially unowned, unnamed
     
133.         semMutex = CreateMutex(0, false, 0);
     
134.         isLocked = false;
     
135. #else
     
136.         int result;
     
137. 
     
138.         memset(&monitor, 0, sizeof(monitor));
     
139.         result = pthread_mutex_init(&monitor, 0);
     
140.         if(result)
     
141.         {
     
142.                 throw runtime_error("Synchronized mutex_init failed!");
     
143.         }
     
144. 
     
145.         memset(&cond, 0, sizeof(cond));
     
146.         result = pthread_cond_init(&cond, 0);
     
147.         if(result)
     
148.         {
     
149.                 throw runtime_error("Synchronized cond_init failed!");
     
150.         }
     
151. #endif
     
152. }
     
153. 
     
154. Synchronized::~Synchronized()
     
155. {
     
156. #ifdef WIN32
     
157.         CloseHandle(semEvent);
     
158.         CloseHandle(semMutex);
     
159. #else
     
160.         int result;
     
161.         result = pthread_cond_destroy(&cond);
     
162. //        if(result)
     
163. //        {
     
164. //                throw runtime_error("Synchronized cond_destroy failed!");
     
165. //        }
     
166.         result = pthread_mutex_destroy(&monitor);
     
167. //        if(result)
     
168. //        {
     
169. //                throw runtime_error("Synchronized mutex_destroy failed!");
     
170. //        }
     
171. #endif
     
172. }
     
173. 
     
174. void Synchronized::wait()
     
175. {
     
176. #ifdef WIN32
     
177.         wait(INFINITE);
     
178. #else
     
179.         cond_timed_wait(0);
     
180. #endif
     
181. }
     
182. 
     
183. #ifndef WIN32
     
184. int Synchronized::cond_timed_wait(const struct timespec *ts)
     
185. {
     
186.   int result;
     
187.   if(ts)
     
188.         result = pthread_cond_timedwait(&cond, &monitor, ts);
     
189.   else
     
190.         result = pthread_cond_wait(&cond, &monitor);
     
191.   return result;
     
192. }
     
193. #endif
     
194. 
     
195. bool Synchronized::wait(unsigned long timeout)
     
196. {
     
197.         bool timeoutOccurred = false;
     
198. #ifdef WIN32
     
199.         isLocked = false;
     
200.         if(!ReleaseMutex(semMutex))
     
201.         {
     
202.                 throw runtime_error("Synchronized: releasing mutex failed");
     
203.         }
     
204.         int err;
     
205.         err = WaitForSingleObject(semEvent, timeout);
     
206.         switch(err)
     
207.         {
     
208.         case WAIT_TIMEOUT:
     
209.                 throw runtime_error("Synchronized: timeout on wait");
     
210.                 timeoutOccurred = true;
     
211.                 break;
     
212.         case WAIT_ABANDONED:
     
213.                 throw runtime_error("Synchronized: waiting for event failed");
     
214.                 break;
     
215.         }
     
216.         if(WaitForSingleObject (semMutex, INFINITE) != WAIT_OBJECT_0)
     
217.         {
     
218.                 throw runtime_error("Synchronized: waiting for mutex failed");
     
219.         }
     
220.         isLocked = true;
     
221. #else
     
222.         struct timespec ts;
     
223.         struct timeval  tv;
     
224.         gettimeofday(&tv, 0);
     
225.         ts.tv_sec  = tv.tv_sec  + (int)timeout/1000;
     
226.         ts.tv_nsec = (tv.tv_usec + (timeout %1000)*1000) * 1000;
     
227. 
     
228.         int err;
     
229.         if((err = cond_timed_wait(&ts)) > 0)
     
230.         {
     
231.                 switch(err)
     
232.                 {
     
233.                 case ETIMEDOUT:
     
234.                   timeoutOccurred = true;
     
235.                   break;
     
236.                 default:
     
237.                   throw runtime_error("Synchronized: wait with timeout returned!");
     
238.                   break;
     
239.                 }
     
240.         }
     
241. #endif
     
242.         return timeoutOccurred;
     
243. }
     
244. 
     
245. void Synchronized::notify()
     
246. {
     
247. #ifdef WIN32
     
248.         numNotifies = 1;
     
249.         if(!SetEvent(semEvent))
     
250.         {
     
251.                 throw runtime_error("Synchronized: notify failed!");
     
252.         }
     
253. #else
     
254.         int result;
     
255.         result = pthread_cond_signal(&cond);
     
256.         if(result)
     
257.         {
     
258.                 throw runtime_error("Synchronized: notify failed!");
     
259.         }
     
260. #endif
     
261. }
     
262. 
     
263. void Synchronized::notify_all()
     
264. {
     
265. #ifdef WIN32
     
266.         numNotifies = (char)0x80;
     
267.         while (numNotifies--)
     
268.         {
     
269.                 if(!SetEvent(semEvent))
     
270.                 {
     
271.                         throw runtime_error("Synchronized: notify_all failed!");
     
272.                 }
     
273.         }
     
274. #else
     
275.         int result;
     
276.         result = pthread_cond_broadcast(&cond);
     
277.         if(result)
     
278.         {
     
279.                 throw runtime_error("Synchronized: notify_all failed!");
     
280.         }
     
281. #endif
     
282. }
     
283. 
     
284. void Synchronized::lock()
     
285. {
     
286. #ifdef WIN32
     
287.     if(WaitForSingleObject(semMutex, INFINITE) != WAIT_OBJECT_0)
     
288.     {
     
289.                 throw runtime_error("Synchronized: lock failed!");
     
290.                 return;
     
291.     }
     
292.     if(isLocked)
     
293.     {
     
294.                 // This thread owns already the lock, but
     
295.             // we do not like recursive locking. Thus
     
296.             // release it immediately and print a warning!
     
297.                 if(!ReleaseMutex(semMutex))
     
298.                 {
     
299.                         throw runtime_error("Synchronized: unlock failed!");
     
300.                 }      
     
301.                 throw runtime_error("Synchronized: recursive locking detected!");
     
302.     }
     
303.     isLocked = true;
     
304. #else
     
305.     pthread_mutex_lock(&monitor);
     
306. #endif
     
307. }
     
308. 
     
309. void Synchronized::unlock()
     
310. {
     
311. #ifdef WIN32
     
312.         isLocked = fasle;
     
313.         if(!ReleaseMutex(semMutex))
     
314.         {
     
315.                 throw runtime_error("Synchronized: unlock failed!");
     
316.                 return;
     
317.         }
     
318. #else
     
319.         pthread_mutex_unlock(&monitor);
     
320. #endif
     
321. }
     
322. 
     
323. bool Synchronized::trylock()
     
324. {
     
325. #ifdef WIN32
     
326.         int status = WaitForSingleObject(semMutex, 0);
     
327.         if(status != WAIT_OBJECT_0)
     
328.         {
     
329.                 throw runtime_error("Synchronized: try lock failed!");
     
330.                 return false;
     
331.         }
     
332.         if(isLocked)
     
333.         {
     
334.                 if(!ReleaseMutex(semMutex))
     
335.                 {
     
336.                         throw runtime_error("Synchronized: unlock failed!");
     
337.                 }
     
338.                 return false;
     
339.         }
     
340.         else
     
341.         {
     
342.                 isLocked = true;
     
343.         }
     
344.         return true;
     
345. #else
     
346.         if(pthread_mutex_trylock(&monitor) == 0)
     
347.                 return true;
     
348.         else
     
349.                 return false;
     
350. #endif
     
351. }
     
352. 
     

复制代码