项目中的数据队列基于轮询和selep的实时性及CUP性能差，需要进行优化，尝试使用concurrentqueue进行优化。网上有一些资料介绍,可供参考。
使用后的个人理解:一个线程安全的queue，并且concurrentqueue的线程安全并不是一味的加锁，它有特殊的技巧，总的来说线程安全且高效但是不保证数据的有序性。是一个很nice的MQ。分析需求，使用阻塞队列，一旦有数据便被读取了，所以使用concurrentqueue可行。
使用：
| 方法 | 功能 | |
|---|---|---|
| ConcurrentQueue(size_t initialSizeEstimate) | 构造函数，它可选地接受队列将包含的元素数量的估计值 | |
| enqueue(T&& item) | 将一个项目排队，必要时分配额外的空间 | |
| try_enqueue(T&& item) | 将一个项目入队，但前提是已经分配了足够的内存,返回值为bool类型 | |
| try_dequeue(T& item) | 将一个项目出队，如果找到一个项目，则返回true；如果队列为空，则返回false | |
| (size_t size_approx() | 返回元素个数 |
| 方法 | 作用 | |
|---|---|---|
| bool try_dequeue_from_producer(producer_token_t const& producer, U& item) | 从特定的生产者队列出列 | |
| size_t try_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max) | 批量出列 | |
| bool is_lock_free() | 检测是否是原子操作 |
封装于：“concurrentqueue.h”
# Allocates more memory if necessary
enqueue(item) : bool
enqueue(prod_token, item) : bool
enqueue_bulk(item_first, count) : bool
enqueue_bulk(prod_token, item_first, count) : bool
# Fails if not enough memory to enqueue
try_enqueue(item) : bool
try_enqueue(prod_token, item) : bool
try_enqueue_bulk(item_first, count) : bool
try_enqueue_bulk(prod_token, item_first, count) : bool
# Attempts to dequeue from the queue (never allocates)
try_dequeue(item&) : bool
try_dequeue(cons_token, item&) : bool
try_dequeue_bulk(item_first, max) : size_t
try_dequeue_bulk(cons_token, item_first, max) : size_t
# If you happen to know which producer you want to dequeue from
try_dequeue_from_producer(prod_token, item&) : bool
try_dequeue_bulk_from_producer(prod_token, item_first, max) : size_t
# A not-necessarily-accurate count of the total number of elements
size_approx() : size_t
简单示例
#include<iostream>
#include"concurrentqueue.h"
int main()
{
int x=0,y=0,z=0;
moodycamel::ConcurrentQueue<int> q;
q.enqueue(2);
q.enqueue(4);
q.enqueue(6);
q.try_dequeue(x);
std::cout << q.size_approx() <<" "<<x<< std::endl;
if (q.is_lock_free)
std::cout << "is lok free" << std::endl;
return 0;
}
#include<iostream>
#include<string>
#include<thread>
#include<mutex>
#include"concurrentqueue.h"
class MintorAbstractBuffer
{
public:
MintorAbstractBuffer(const uint32_t buffersize)
:_Buffersize(buffersize)
{}
bool push(std::string s)
{
int32_t size = _buffer.size_approx();
if (_Buffersize > size)
{
if (_buffer.enqueue(s))
{
++_Size;
return true;
}
}
return false;
}
bool RecursivePush(std::string s,int32_t num)
{
//std::lock_guard<std::mutex> lock(_mut);
for (int32_t i = 0; i < num; ++i)
{
int32_t size = _buffer.size_approx();
if (size >= _Buffersize)
{
std::cout << "Push Error" << " ";
return false;
}
_buffer.enqueue(s);
++_Size;
}
return true;
}
void pint()
{
int32_t size = _buffer.size_approx();
for (int32_t i = 0; i <= size; ++i)
{
std::string curs;
if (!_buffer.try_dequeue(curs))
std::cout << "Pint Error" << std::endl;
std::cout << curs<< i <<" ";
if (!_buffer.enqueue(curs))
std::cout << "Pint Error" << std::endl;
}
std::cout << std::endl;
std::cout <<_buffer.size_approx()<< std::endl;
std::cout <<_Size<< std::endl;
}
private:
std::mutex _mut;
int32_t _Buffersize;
moodycamel::ConcurrentQueue<std::string> _buffer;
uint32_t _Size=0;
};
int main()
{
std::string s1 = "a", s2 = "b", s3 = "c";
MintorAbstractBuffer buffer(100000);
std::thread t1(&MintorAbstractBuffer::RecursivePush, &buffer, s1, 333);
std::thread t2(&MintorAbstractBuffer::RecursivePush, &buffer, s2, 333);
std::thread t3(&MintorAbstractBuffer::RecursivePush, &buffer, s3, 334);
t1.join();t2.join();t3.join();
buffer.pint();
return 0;
}
加锁之后使得++size不会被打断：
_size是cont的push次数，在++size过程中可能被其他线程的语句终止导致并没有++size，大概是1％的终止率。
想要数据有序还是要使用STL::queue，或者使用ConcurrentQueue的阻塞版本，见文末
#include<iostream>
#include<string>
#include<thread>
#include<mutex>
#include<queue>
class MintorAbstractBuffer
{
public:
MintorAbstractBuffer(const uint32_t buffersize)
:_Buffersize(buffersize)
{}
bool push(std::string s)
{
int32_t size = _buffer.size();
if (_Buffersize > size)
{
_buffer.push(s);
++_Size;
return true;
}
return false;
}
bool RecursivePush(std::string s,int32_t num)
{
std::lock_guard<std::mutex> lock(_mut);
for (int32_t i = 0; i < num; ++i)
{
int32_t size = _buffer.size();
if (size > _Buffersize)
{
std::cout << "Push Error" << " ";
return false;
}
_buffer.push(s);
++_Size;
}
return true;
}
void pint()
{
int32_t size = _buffer.size();
for (int32_t i = 0; i <= size; ++i)
{
std::cout << i;
std::cout << _buffer.front() << " ";
std::string curs;
curs = _buffer.front();
_buffer.pop();
_buffer.push(curs);
}
std::cout <<std::endl;
std::cout << "push cont ：" << _Size << std::endl;
std::cout << "_buffer.size() ：" << _buffer.size() << std::endl;
}
private:
std::mutex _mut;
int32_t _Buffersize;
std::queue<std::string> _buffer;
uint32_t _Size=0;
};
int main()
{
std::string s1 = "a", s2 = "b", s3 = "c";
MintorAbstractBuffer buffer(100000);
int32_t x=3333;
std::thread t1(&MintorAbstractBuffer::RecursivePush, &buffer, s1, x);
std::thread t2(&MintorAbstractBuffer::RecursivePush, &buffer, s2, x);
std::thread t3(&MintorAbstractBuffer::RecursivePush, &buffer, s3, x+1);
t1.join();t2.join();t3.join();
buffer.pint();
return 0;
}
封装于：“blockingconcurrentqueue.h”，在工程中使用一个线程不断使用阻塞版本API调用数据可以提升数据的处理速度(缓冲区基本为空)。当然需要知道阻塞与非阻塞队列的区别，下面的API可以通过调整参数来控制多长时间从阻塞的队列中重新获取数据。github上有详细的介绍,出队操作有更改API如下。
wait_dequeue(U& item) : void
wait_dequeue_timed(U& item, std::int64_t timeout_usecs) : bool //时间参数 std::chrono::milliseconds(5)
wait_dequeue_bulk(It itemFirst, size_t max) : size_t
wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::int64_t timeout_usecs) : size_t