“生产者-消费者”模型的一个小问题。

fender0107401

问题：生产者在把message压入message_buffer的时候，使用“阻塞”或者“非阻塞”模式都没什么区别吧？

目标：我想在每次压入message_buffer的时候都单独打开一个thread，然后让这个新线程来负责具体的压入工作，这样生产者就变成了“非阻塞”模式。

反正就是想尽力利用系统的多核心来提高处理速度。

windoze

你的message queue有尺寸上限么？没有就没关系，反正push queue也不会block。

但是“在每次压入message_buffer的时候都单独打开一个thread”这句我有点理解不能。

fender0107401

回复 2# windoze


意思是说我想这样：boost::thread o_thread((boost::bind(&message_buffer::put_operation, this, p_message)));

1. #include <queue>
   
2. #include <boost/shared_ptr.hpp>
   
3. #include "message.hpp"
   
4. 
   
5. class message_buffer: public boost::noncopyable
   
6. {
   
7.     private:
   
8. 
   
9.         boost::mutex                  o_mutex;
   
10.         boost::condition_variable_any o_get_cond;
    
11. 
    
12.         std::queue<boost::shared_ptr<message>> o_queue;
    
13. 
    
14.         int current_size = 0;
    
15. 
    
16.     public:
    
17.         void put(boost::shared_ptr<message> p_message)
    
18.         {
    
19.             boost::thread o_thread((boost::bind(&message_buffer::put_operation, this, p_message)));
    
20.         }
    
21. 
    
22.         void get(boost::shared_ptr<message> p_message)
    
23.         {
    
24.             mutex::lock_guard o_lock_guard(this->o_mutex);
    
25. 
    
26.             while (this->is_empty())
    
27.             {
    
28.                 this->o_get_cond.wait(this->o_mutex);
    
29.             }
    
30. 
    
31.             this->current_size--;
    
32. 
    
33.             p_message = *(this->o_queue.front());
    
34. 
    
35.             this->o_queue.pop();
    
36.         }
    
37. 
    
38.     private:
    
39.         bool is_empty(void)
    
40.         {
    
41.             return this->current_size == 0;
    
42.         }
    
43. 
    
44.         void put_operation(boost::shared_ptr<message> p_message)
    
45.         {
    
46.             {
    
47.                 mutex::lock_guard o_lock_guard(this->o_mutex);
    
48. 
    
49.                 this->o_queue.push(p_message);
    
50. 
    
51.                 this->current_size++;
    
52.             }
    
53. 
    
54.             this->o_get_cond.notify_one()
    
55.         }
    
56. }
    
57. 
    

复制代码

fender0107401

没怎么弄过这个，不知道这样行不行，我正在冥思苦想各种提升性能（发挥硬件性能）的方法。

fender0107401

突然发现个问题：如果queue的push操作不会block，那我就不用这么折腾了，直接就push就行了。

果然学艺不精。。。

windoze

回复 5# fender0107401

就算push会block你也不用折腾啊，无非就是加一个full_condition在push的时候wait就好了。

不过有个问题，你这个message_buffer怎么关闭？

fender0107401

回复 6# windoze

1.我不想wait，我就想生产者直接put一下，然后put这个函数打开一个新的线程以后就返回（一点都不想等，我担心producer在put的时候正好有几个consumer在等着get会影响producer的速度），然后这个线程把message给push进去。

2.“不过有个问题，你这个message_buffer怎么关闭？”这句我不太明白，关闭是什么意思？


   

wlmqgzm

高性能"lock-free"无锁队列的应用:  boost::lockfree::spsc_queue  一个无等待的单生产者/单消费者队列
2个线程之间, 必须要交换数据的问题,  其中一个线程是管理线程, 一个是acceptor线程.
对于此类问题, 传统的方式就是: 生产者和消费者模式, 二者之间, 通过队列QUEUE, 双方都加锁(mutex)处理队列.
我的解决方案是采用  Boost 高性能 无锁队列来实现,  只用了三条代码就解决了此问题.
无锁队列的优势:
它将保证线程无限次调用这个方法都能够在有限步内完成，而不会因为其他线程被阻塞而导致本线程无法在有限步内完成，即“无锁”。
保证在任何情况下所有并发中有一个操作能够得到执行, 不必考虑"多线进程的优先级反转等策略".

#include <boost/lockfree/spsc_queue.hpp>
boost::lockfree::spsc_queue<unsigned int, boost::lockfree::capacity<65536> >  closed_ip_queue;  
closed_ip_queue.push( int_ip );
closed_ip_queue.pop( int_closed_ip )
还要加上检查返回值的代码

下面是Boost对于spsc_queue的解释:
boost::lockfree::spsc_queue
await-free single-producer/single-consumer queue (commonly known as ringbuffer)
一个无等待的单生产者/单消费者队列（通常被称为环形缓冲区）  
这个非常猛的东西, 号称无锁, 无等待, 双并发.......

fender0107401

回复 8# wlmqgzm

我刚刚在网上找到了这个，http://www.boost.org/doc/libs/1_ ... lockfree/queue.html

你发的这个是<boost/lockfree/spsc_queue.hpp>，我找到那个是<boost/lockfree/queue.hpp>。

我感觉用这个以后，就不用自己写P/C模型了。

wlmqgzm

回复 9# fender0107401

这个我用了多年了.   记得大约6-12个月前看windoze的开源代码, 有几个队列,  里面一堆的锁, 然后给他回复, 粘贴了Boost手册上的例子, 没有被采纳,

另外, Boost手册 lock-free queue的例子有错误, 在直接拿来测试的时候, 编译应该不会通过, 要自己修改,

这个也是多数人实验, 就编译错误, 然后放弃这个库, 因此lock-free queue在国内使用不多的原因之一,

修改方法,参考我的三行代码.  里面要增补定义boost::lockfree::capacity<XXX>

#include <boost/lockfree/spsc_queue.hpp>
boost::lockfree::spsc_queue<unsigned int, boost::lockfree::capacity<65536> >  closed_ip_queue;  
closed_ip_queue.push( int_ip );
closed_ip_queue.pop( int_closed_ip )
还要加上检查返回值的代码