阻塞队列实现

0 背景

阻塞队列在生产业务的很多场景里，都有使用需要。例如：在数据流式处理服务中，我们需要并行化的运行上游逻辑a与下游逻辑b；在rpc网络通讯框架场景中，我们需要解耦网络读写线程与消息处理线程。总之,阻塞队列是在不同职责的线程之间进行数据沟通的桥梁。在java的jdk里面有现成的BlockingQueue类，让我们看看一个C++版的实现吧：

1 实现源码

1.1 锁封装

通过信号量保证不同线程之间数据操作的一致性。

class CSemaphore

{

 public:

  CSemaphore()

  {

    sem_init(&m_sem, 0, 0);

  }

  ~CSemaphore()

  {

    sem_destroy(&m_sem);

  }

  void Produce()

  {

    sem_post(&m_sem);

  }

  void Consume()

  {

    while (sem_wait(&m_sem) != 0)

    {

      sched_yield();

    }

  }

  bool Try()

  {

    int value = 0;

    int ret = sem_getvalue(&m_sem, &value);

    if(ret < 0 || value <= 0)

    {

      return false;

    }

    return true;

  }

  bool TryTime(int micSec)

  {

    struct timespec ts;

    clock_gettime(CLOCK_REALTIME,&ts);

    if(micSec >= 1000000)

      ts.tv_sec += micSec/1000000;

    ts.tv_nsec += micSec%1000000*1000;

    if(ts.tv_nsec >= 1000000000)

    {

      ++ts.tv_sec;

      ts.tv_nsec -= 1000000000;

    }

    int ret = sem_timedwait(&m_sem,&ts);

    if(ret < 0)

      return false;

    return true;

  }

  int GetCount()

  {

    int value = 0;

    int ret = sem_getvalue(&m_sem, &value);

    if(ret < 0)

      return -1;

    else

      return value;

  }

 private:

  sem_t  m_sem;

};

1.2 写数据(消息)

  bool Put(const __T& value)

  {

    CMutexLock lock(m_mutex);

    if(m_queue.size() > m_queMax)

    { 

      return false;

    } 

    m_queue.push_back(value);

    m_semaphore.Produce();

    return true;

  }

1.3 读取数据(消息)

  bool Get(__T& value)

  {

    m_semaphore.Consume();

    CMutexLock lock(m_mutex);

    if(m_queue.empty())

      return false;

    value = m_queue.front();

    m_queue.pop_front();

    return true;

  }

2 实例

2.1 源码

2.2 输出

源码请见附件http://files.cnblogs.com/files/gisorange/blockqueue.zip