// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
//
// Author: Shuo Chen (chenshuo at chenshuo dot com)#ifndef MUDUO_BASE_BOUNDEDBLOCKINGQUEUE_H
#define MUDUO_BASE_BOUNDEDBLOCKINGQUEUE_H#include <muduo/base/Condition.h>
#include <muduo/base/Mutex.h>#include <boost/circular_buffer.hpp>
#include <boost/noncopyable.hpp>
#include <assert.h>namespace muduo
{template<typename T>
class BoundedBlockingQueue : boost::noncopyable
{public:explicit BoundedBlockingQueue(int maxSize): mutex_(),notEmpty_(mutex_),notFull_(mutex_),queue_(maxSize){}void put(const T& x){MutexLockGuard lock(mutex_);while (queue_.full()){notFull_.wait();}assert(!queue_.full());queue_.push_back(x);notEmpty_.notify(); // TODO: move outside of lock}T take(){MutexLockGuard lock(mutex_);while (queue_.empty()){notEmpty_.wait();}assert(!queue_.empty());T front(queue_.front());queue_.pop_front();notFull_.notify(); // TODO: move outside of lockreturn front;}bool empty() const{MutexLockGuard lock(mutex_);return queue_.empty();}bool full() const{MutexLockGuard lock(mutex_);return queue_.full();}size_t size() const{MutexLockGuard lock(mutex_);return queue_.size();}size_t capacity() const{MutexLockGuard lock(mutex_);return queue_.capacity();}private:mutable MutexLock          mutex_;Condition                  notEmpty_;Condition                  notFull_;boost::circular_buffer<T>  queue_;
};}#endif  // MUDUO_BASE_BOUNDEDBLOCKINGQUEUE_H

#include <muduo/base/BlockingQueue.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Thread.h>
#include <muduo/base/Timestamp.h>#include <boost/bind.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <map>
#include <string>
#include <stdio.h>class Bench
{public:Bench(int numThreads): latch_(numThreads),threads_(numThreads){for (int i = 0; i < numThreads; ++i){char name[32];snprintf(name, sizeof name, "work thread %d", i);threads_.push_back(new muduo::Thread(boost::bind(&Bench::threadFunc, this), muduo::string(name)));}for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::start, _1));}void run(int times){printf("waiting for count down latch\n");latch_.wait();printf("all threads started\n");for (int i = 0; i < times; ++i){muduo::Timestamp now(muduo::Timestamp::now());queue_.put(now);usleep(1000);}}void joinAll(){for (size_t i = 0; i < threads_.size(); ++i){queue_.put(muduo::Timestamp::invalid());}for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::join, _1));}private:void threadFunc(){printf("tid=%d, %s started\n",muduo::CurrentThread::tid(),muduo::CurrentThread::name());std::map<int, int> delays;latch_.countDown();bool running = true;while (running){muduo::Timestamp t(queue_.take());muduo::Timestamp now(muduo::Timestamp::now());if (t.valid()){int delay = static_cast<int>(timeDifference(now, t) * 1000000);// printf("tid=%d, latency = %d us\n",//        muduo::CurrentThread::tid(), delay);++delays[delay];}running = t.valid();}printf("tid=%d, %s stopped\n",muduo::CurrentThread::tid(),muduo::CurrentThread::name());for (std::map<int, int>::iterator it = delays.begin();it != delays.end(); ++it){printf("tid = %d, delay = %d, count = %d\n",muduo::CurrentThread::tid(),it->first, it->second);}}muduo::BlockingQueue<muduo::Timestamp> queue_;muduo::CountDownLatch latch_;boost::ptr_vector<muduo::Thread> threads_;
};int main(int argc, char* argv[])
{int threads = argc > 1 ? atoi(argv[1]) : 1;Bench t(threads);t.run(10000);t.joinAll();
}

2.BlockinngQueue<T>（无界缓冲区）

// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
//
// Author: Shuo Chen (chenshuo at chenshuo dot com)#ifndef MUDUO_BASE_BLOCKINGQUEUE_H
#define MUDUO_BASE_BLOCKINGQUEUE_H#include <muduo/base/Condition.h>
#include <muduo/base/Mutex.h>#include <boost/noncopyable.hpp>
#include <deque>
#include <assert.h>namespace muduo
{template<typename T>
class BlockingQueue : boost::noncopyable
{public:BlockingQueue(): mutex_(),notEmpty_(mutex_),queue_(){}void put(const T& x){MutexLockGuard lock(mutex_);queue_.push_back(x);notEmpty_.notify(); // TODO: move outside of lock}T take(){MutexLockGuard lock(mutex_);// always use a while-loop, due to spurious wakeupwhile (queue_.empty()){notEmpty_.wait();}assert(!queue_.empty());T front(queue_.front());queue_.pop_front();return front;}size_t size() const{MutexLockGuard lock(mutex_);return queue_.size();}private:mutable MutexLock mutex_;Condition         notEmpty_;std::deque<T>     queue_;
};}#endif  // MUDUO_BASE_BLOCKINGQUEUE_H

#include <muduo/base/BlockingQueue.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Thread.h>#include <boost/bind.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <string>
#include <stdio.h>class Test
{public:Test(int numThreads): latch_(numThreads),threads_(numThreads){for (int i = 0; i < numThreads; ++i){char name[32];snprintf(name, sizeof name, "work thread %d", i);threads_.push_back(new muduo::Thread(boost::bind(&Test::threadFunc, this), muduo::string(name)));}for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::start, _1));}void run(int times){printf("waiting for count down latch\n");latch_.wait();printf("all threads started\n");for (int i = 0; i < times; ++i){char buf[32];snprintf(buf, sizeof buf, "hello %d", i);queue_.put(buf);printf("tid=%d, put data = %s, size = %zd\n", muduo::CurrentThread::tid(), buf, queue_.size());}}void joinAll(){for (size_t i = 0; i < threads_.size(); ++i){queue_.put("stop");}for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::join, _1));}private:void threadFunc(){printf("tid=%d, %s started\n",muduo::CurrentThread::tid(),muduo::CurrentThread::name());latch_.countDown();bool running = true;while (running){std::string d(queue_.take());printf("tid=%d, get data = %s, size = %zd\n", muduo::CurrentThread::tid(), d.c_str(), queue_.size());running = (d != "stop");}printf("tid=%d, %s stopped\n",muduo::CurrentThread::tid(),muduo::CurrentThread::name());}muduo::BlockingQueue<std::string> queue_;muduo::CountDownLatch latch_;boost::ptr_vector<muduo::Thread> threads_;
};int main()
{printf("pid=%d, tid=%d\n", ::getpid(), muduo::CurrentThread::tid());Test t(5);t.run(100);t.joinAll();printf("number of created threads %d\n", muduo::Thread::numCreated());
}