C++ 无锁队列实现

上源码

#ifndef __GLOBAL_LOCK_FREE_QUEUE_H__
#define __GLOBAL_LOCK_FREE_QUEUE_H__

#include <atomic>
#include <list>

#ifdef _WINDOWS
#include <windows.h>
//PVOID __cdecl InterlockedCompareExchangePointer(
//    _Inout_ PVOID volatile *Destination,
//    _In_    PVOID          Exchange,
//    _In_    PVOID          Comparand
//    );
//
//Parameters
//    Destination [in, out]
//A pointer to a pointer to the destination value.
//    Exchange [in]
//The exchange value.
//    Comparand [in]
//The value to compare to Destination.
//    Return value
//    The function returns the initial value of the Destination parameter.
//    Remarks
//    The function compares the Destination value with the Comparand value. If the Destination value is equal to the Comparand value, the Exchange value is stored in the address specified by Destination. Otherwise, no operation is performed.
//    On a 64-bit system, the parameters are 64 bits and must be aligned on 64-bit boundaries; otherwise, the function will behave unpredictably. On a 32-bit system, the parameters are 32 bits and must be aligned on 32-bit boundaries.
//    The interlocked functions provide a simple mechanism for synchronizing access to a variable that is shared by multiple threads. This function is atomic with respect to calls to other interlocked functions.
#define __sync_bool_compare_and_swap(a,b,c) (InterlockedCompareExchangePointer((void*volatile*)a,c,b), (*a)==(c))
#endif

namespace DataCache
{

    template <typename T>
    class LinkList
    {
    public:
        T data;
        LinkList<T> *next;
    };// class LinkeList<T>

    template <typename T>
    class LockFreeQueue
    {
    public:
        LockFreeQueue();

        void push_back(T t);

        T pop_front(void);

        bool isEmpty(void);

        int GetLength();

    private:
        LinkList<T> *head_;
        LinkList<T> *tail_;
        std::_Atomic_integral_t elementNumbers_;
    }; // class LockFreeQueue

    template <typename T>
    LockFreeQueue<T>::LockFreeQueue()
        :head_(NULL),
        tail_(new LinkList<T>),
        elementNumbers_(0)
    {
        head_ = tail_;
        tail_->next = NULL;
    }

    template <typename T>
    void LockFreeQueue<T>::push_back(T t)
    {
        auto newVal = new LinkList<T>;
        newVal->data = t;
        newVal->next = NULL;

        LinkList<T> *p;
        do
        {
            p = tail_;
        } while (!__sync_bool_compare_and_swap(&tail_->next, NULL, newVal));

        //move tail_
        __sync_bool_compare_and_swap(&tail_, tail_, newVal);
        elementNumbers_++;
    }

    template <typename T>
    T LockFreeQueue<T>::pop_front()
    {
        LinkList<T> *p;

        do
        {
            //record the first node.
            p = head_->next;
            if (!p)
            {
                return 0;
            }
        } while (!__sync_bool_compare_and_swap(&head_->next, p, p->next));

        if (elementNumbers_ > 0) elementNumbers_--;
        if (elementNumbers_ == 0)
        {
            // if the queue is empty then the tail to header.
            do
            {

            } while (!__sync_bool_compare_and_swap(&tail_, p, head_));
        }

        return p->data;
    }

    template <typename T>
    bool LockFreeQueue<T>::isEmpty()
    {
        if (elementNumbers_ == 0)
        {
            return true;
        }
        else
        {
            return false;
        }
    }

    template <typename T>
    int LockFreeQueue<T>::GetLength()
    {
        return elementNumbers_;
    }

}// namespace DataCache

#endif

 

 

源码解析：

重点在函数InterlockedCompareExchangePointer(a,c,b),其作用为如果a==b,则修改a的值为c，这是微软提供的一个原子操作的API，这里面通过定义这个宏，来进行queue入队列和出队列时的指针修改，达到无锁操作的目的，可以大幅提高队列的操作性能，避免过程中进行加锁操作。在GCC中已经原生支持了__sync_bool_compare_and_swap，所以这个宏的作用是针对MSVC版本提供的类似函数。

 

测试代码：

#include"lockFreeQueue.h"

DataCache::LockFreeQueue<int> queue;

void func1(void)
{
    for(int i = 0; i <100000 ; i++)
    {
        queue.push_back(i);
    }
}

void func2(void)
{
    for(int i = 100000; i < 200000; i++)
    {
        queue.push_back(i);
    }
}

void func5(void)
{
    for(int i = 200000; i < 300000; i++)
    {
        queue.push_back(i);
    }
}

void func4(void)
{
    for(int i = 400000; i < 500000; i++)
    {
        queue.push_back(i);
    }
}

void func3(void)
{
    while (!queue.isEmpty())
    {
        std::cout<<queue.pop_front()<<" ";
    }

    std::cout<<std::endl;
}

int main(int argc,char **argv)
{
    std::thread t1(func1);
    std::thread t2(func2);
    std::thread t5(func5);
    std::thread t6(func4);

    t1.join();
    t2.join();
    t5.join();
    t6.join();

    int length=queue.GetLength();
    assert(length==500000);
    return 0;
}