多线程问题

youzlm

需求：
一个队列，刚开始为空
一些线程给队列填数据。
别一些线程读队列数据。
问题：
我想专门开几个线程来等待队列不是空时就取出来显示。
要如何监听这个队列，来唤醒线程读取数据

hellioncu

条件变量

teng0210

semphore

jkh7

读线程：pthread_cond_wait
写线程：pthread_cond_broadcast

youzlm

读线程一个，写进程多个

那是不是每写一个数据都要唤醒读线程吗？

youzlm

在读的线程放一个pthread_cond_wait，
在写线程上放pthread_cond_broadcast，
当读线程上有数据就入进队更，唤醒写线程。
这样变成了，队列没有用，一次只处理一个数据

如何改一下。。就是写线程一有数据就写进队列。
读线程只要队列有数据就读数据

wwwhenan

不知道是不是你的需求。

1. 
   
2. /********************************************************************
   
3. created:        2009/10/08
   
4. modify :    2009/10/08
   
5. file :         TQueue.h
   
6. author:               
   
7. purpose:    Thread Safe FIFO queue
   
8. *********************************************************************/
   
9. 
   
10. #ifndef TQUEUE_H_
    
11. #define TQUEUE_H_
    
12. 
    
13. #include "apr_thread_mutex.h"
    
14. #include "apr_thread_cond.h"
    
15. 
    
16. #include <deque>
    
17. #include <assert.h>
    
18. 
    
19. ////////////////////////////////////////////////////////////////////////////
    
20. namespace comn
    
21. {
    
22.         template <class A>
    
23.         class CQueue
    
24.         {
    
25.         public:               
    
26.                 CQueue();       
    
27.                 ~CQueue();
    
28. 
    
29.                 ///压入一个元素
    
30.                 void push(const A &item);                                        //for producer
    
31. 
    
32.                 bool pop(A &item);                                                        //for consumer
    
33. 
    
34.                 bool popTimeWait(A &item,int millisec);     //for consumer
    
35. 
    
36.                 size_t  size();     //队列大小
    
37.                 bool    empty();    ///判为空
    
38. 
    
39.                 void    post(); ///激活资源获取者
    
40.        
    
41.         private:               
    
42.                 std::deque<A> m_queue;    ///stl deque
    
43. 
    
44.                 apr_pool_t *m_aprBasePool;
    
45.                 apr_thread_mutex_t *m_mutex;   ///线程锁
    
46.                 apr_thread_cond_t *m_cond;     ///条件变量
    
47.         };
    
48. 
    
49. 
    
50.         template <class A>
    
51.         CQueue<A>::CQueue()
    
52.         {       
    
53.                 apr_status_t ret;
    
54.                 ret = apr_pool_create(&m_aprBasePool,NULL);               
    
55.                 if (ret != APR_SUCCESS)
    
56.                 {
    
57.                         assert(0);
    
58.                 }
    
59. 
    
60.                 ///线程锁
    
61.                 ret = apr_thread_mutex_create(&m_mutex,APR_THREAD_MUTEX_DEFAULT,m_aprBasePool);       
    
62.                 if (ret != APR_SUCCESS)
    
63.                 {
    
64.                         assert(0);
    
65.                 }
    
66. 
    
67.                 ///条件变量
    
68.                 ret = apr_thread_cond_create(&m_cond, m_aprBasePool);       
    
69.                 if (ret != APR_SUCCESS)
    
70.                 {
    
71.                         assert(0);
    
72.                 }
    
73.                 m_queue.clear();
    
74.         }
    
75. 
    
76. 
    
77. 
    
78.         template <class A>
    
79.         CQueue<A>::~CQueue()
    
80.         {
    
81.                 m_queue.clear();
    
82.                 apr_thread_mutex_destroy(m_mutex);
    
83.                 apr_thread_cond_destroy(m_cond);
    
84.                 apr_pool_destroy(m_aprBasePool);
    
85.         }
    
86. 
    
87.         template <class A>
    
88.         void CQueue<A>::push(const A &item)
    
89.         {
    
90.                 apr_thread_mutex_lock(m_mutex);                               
    
91.                 m_queue.push_back(item);        ///数据放入队列
    
92.                 apr_thread_mutex_unlock(m_mutex);
    
93. 
    
94.                 //唤醒一个线程
    
95.                 apr_thread_cond_signal(m_cond);   
    
96.                 return ;
    
97.         }
    
98. 
    
99. 
    
100.         template <class A>
     
101.         void CQueue<A>::post()
     
102.         {       
     
103.                 apr_thread_cond_signal(m_cond);
     
104.         }
     
105. 
     
106. 
     
107.         template <class A>
     
108.         size_t CQueue<A>::size()
     
109.         {
     
110.                 apr_thread_mutex_lock(m_mutex);       
     
111.                 size_t s = m_queue.size();
     
112.                 apr_thread_mutex_unlock(m_mutex);       
     
113.                 return s;
     
114.         }
     
115. 
     
116.         template <class A>
     
117.         bool CQueue<A>::empty()
     
118.         {               
     
119.                 return size() == 0;
     
120.         }
     
121. 
     
122.         template <class A>
     
123.         bool CQueue<A>::pop(A &item)
     
124.         {       
     
125.                 apr_thread_mutex_lock(m_mutex);       
     
126.                 while (m_queue.empty())
     
127.                 {       
     
128.                         //线程睡眠解锁，等待唤醒
     
129.                         apr_thread_cond_wait(m_cond,m_mutex);  
     
130.                 }       
     
131. 
     
132.                 if(m_queue.empty())
     
133.                 {       
     
134.                         apr_thread_mutex_unlock(m_mutex);
     
135.                         return false;
     
136.                 }
     
137. 
     
138.                 item = m_queue.front();  ///取头元素
     
139.                 m_queue.pop_front();     ///弹出头元素       
     
140. 
     
141.                 apr_thread_mutex_unlock(m_mutex);
     
142.                 return true;
     
143.         }
     
144. 
     
145.         template <class A>
     
146.         bool CQueue<A>::popTimeWait(A &item,int millisec)
     
147.         {       
     
148.                 apr_thread_mutex_lock(m_mutex);
     
149. 
     
150.                 while (m_queue.empty())
     
151.                 {               
     
152.                         //线程睡眠解锁，等待唤醒或者超时
     
153.                         apr_status_t ret = apr_thread_cond_timedwait(m_cond,m_mutex,millisec);
     
154. 
     
155.                         if (ret == APR_TIMEUP)      ///超时了
     
156.                         {
     
157.                                 apr_thread_mutex_unlock(m_mutex);
     
158.                                 return false;
     
159.                         }
     
160.                 }               
     
161. 
     
162.                 item = m_queue.front();  ///取头元素
     
163.                 m_queue.pop_front();       
     
164. 
     
165.                 apr_thread_mutex_unlock(m_mutex);
     
166. 
     
167.                 return true;
     
168.         }
     
169. 
     
170. }
     
171. ////////////////////////////////////////////////////////////////////////////
     
172. #endif /* TQUEUE_H_ */
     

复制代码

ddgfff

多线程互斥锁使用实现：信号量 Semaphore


入队列input信号量 +1
出队列output信号量 -1