C++11 – 基于无锁队列的单生产者单消费者模型

1 单生产者单消费者无锁队列

Github项目地址：https://github.com/cameron314/readerwriterqueue

1.1 成员函数说明

• try_enqueue：如果队列中有空间，则使元素的副本入队。如果元素已入队，则返回true，否则返回false。不分配内存。
• enqueue ： 使元素的副本进入队列。如果需要，分配一个额外的内存块。仅在内存分配失败时失败（返回false）。
• try_dequeue ： 尝试使元素出队； 如果队列为空，返回false。 如果队列中至少有一个元素，使用operator =将结果移到结果的前面，然后返回true。
• peek ： 返回一个指向队列中前元素的指针（接下来将通过调用try_dequeue或pop删除）。 如果
  调用该方法时，队列显示为空，返回nullptr。必须仅从消费者线程调用。
• pop ： 从队列中删除前元素（如果有），而不返回它。成功返回true，如果队列当时为空则返回false。
• size_approx ：返回队列中当前的大概项目数。从生产者线程和使用者线程都可以安全调用。
• max_capacity ： 返回此队列为空时可排队但不分配的项目总数。 生产者线程和消费者线程均可安全调用。
• emplace ： 类似于enqueue（），但具有Emplace语义（就地构造）
• try_emplace ： 类似于try_enqueue（），但具有Emplace语义（就地构造）

1.2 使用

在项目中包含readerwriterqueue.h和atomicops.h

2 多生产者多消费者无锁队列

另外这个作者还搞了一个并发的无锁队列，
Github地址：https://github.com/cameron314/concurrentqueue

3 单生产者单消费者无锁模型

基于第一个项目的readerwriterqueue，做了以下代码测试：

#include <iostream>
#include <thread>

#include "readerwriterqueue.h"

using namespace moodycamel;


ReaderWriterQueue<int> q(100);

int globalCount = 0;
// 生产者线程
void Producer()
{
    while (true)
    {
        globalCount++;
        q.enqueue(globalCount);
    }
}

void Consumer()
{
    while (true)
    {
        int temp;
        q.try_dequeue(temp);
        std::cout <<temp << std::endl;
    }
}



int main()
{
    std::thread consumer_Thread(Consumer);
    std::thread producer_Thread(Producer);

    consumer_Thread.join();
    producer_Thread.join();

    getchar();
    return 0;
}

速度是真的快，果然如这个作者说的：最快的同步就是永不不会发生的同步。而且其压出队列的顺序做出了保证。

4 如果队列里面是指针，该如何正确释放内存

可参考以下代码：

#include <iostream>
#include <thread>
#include <vld.h>

#include "readerwriterqueue.h"
#include "concurrentqueue.h"

using namespace moodycamel;

class A {
public:
    int a;
};

ReaderWriterQueue<A*> q;

// 生产者线程
void Producer()
{
    for (int i = 0; i < 100000; ++i)
    {
        A* temp = new A();
        temp->a = i;
        q.enqueue(temp);
    }
}

void Consumer()
{
    while (true)
    {
        A* temp = NULL;
        if (q.try_dequeue(temp))
        {
            //std::cout << temp->a <<"队列数："<<q.size_approx()<< std::endl;
            delete temp;
            temp = NULL;
        }
        else
        {
            std::cout << "执行完成" << std::endl;
            break;
        }   

    }
}



int main()
{
    Producer();
    std::cout << "开始执行" << std::endl;
    std::thread consumer_Thread(Consumer);

    consumer_Thread.join();

    getchar();
    return 0;
}