无锁编程[2]__一读一写无锁队列，其实就是循环队列，一写多读(基于计数器和更新开关)，基于CAS实现：多读多写无锁循环队列

转自：http://blog.163.com/xychenbaihu@yeah/blog/static/13222965520133794137729/

一、数组循环队列：一读一写无锁循环队列。

//message_queue.h

typedef struct MsgNode

{

unsigned flag;

void* content;

}MsgNode;

class CMessageQueue //循环队列

{

public:

CMessageQueue(int size = 8192,int fd = -1);

~CMessageQueue();

//无锁添加和取消息

int send_message(unsigned msg,void* content);

int get_message(MsgNode& msg);

private:

MsgNode* msgbuf;

int max_msg_size;

int head_pos;

int tail_pos;

};

//message_queue.cpp

CMessageQueue::CMessageQueue(int size,int fd)

{

msgbuf = new MsgNode[size];

memset(msgbuf,fd,sizeof(MsgNode)*size);

max_msg_siz e= size;

head_pos = 0;

tail_pos = 0;

}

CMessageQueue::~CMessageQueue(int size,int fd)

{

delete [] msgbuf;

max_msg_size = 0;

head_pos = 0;

tail_pos = 0;

}

int CMessageQueue::send_message(unsigned msg,void* content)

{

MsgNode* pObj = &msg_buf[tail_pos];

if(pObj->flag != (unsigned)-1) //最后一个节点被使用了

{

//队列满

return -1;

}

tail_pos=(tail_pos + 1)%max_msg_size；

//申请一个消息记录,将消息记录赋值

pObj->content = content;

pObj->flag = msg; //此字段放最后赋值

return 0;

}

int CMessageQueue::get_message(MsgNode& msg)

{

MsgNode* pObj = &msg_buf[head_pos];

if(pObj->flag == (unsigned)-1)

{

//队列为空

return -1;

}

head_pos=(head_pos+1)%max_msg_size;

//重置消息

msg = *pObj;

pObj->Msg = (unsigned)-1;

return 0;

}

模版数组循环队列的实现可以参考：

http://blog.163.com/xychenbaihu@yeah/blog/static/13222965520132343915442/

二、数组循环队列：多读多写无锁循环队列，使用CAS。

do

{

...

}

while(cas) 这里可以用sched_yield();让出cpu。

//message_queue.h

//异步消息

typedef struct MsgNode

{

unsigned flag;

void* content;

}MsgNode;

class CMessageQueue //循环队列

{

public:

CMessageQueue(int size = 8192,int fd = -1);

~CMessageQueue();

//无锁添加和取消息

int send_message(unsigned msg,void* content);

int get_message(MsgNode& msg);

private:

MsgNode* msgbuf;

int max_msg_size;

int head_pos;

int tail_pos;

};

//message_queue.cpp

CMessageQueue::CMessageQueue(int size,int fd)

{

msgbuf = new MsgNode[size];

memset(msgbuf,fd,sizeof(MsgNode)*size);

max_msg_siz e= size;

head_pos = 0;

tail_pos = 0;

}

CMessageQueue::~CMessageQueue(int size,int fd)

{

delete [] msgbuf;

max_msg_size = 0;

head_pos = 0;

tail_pos = 0;

}

int CMessageQueue::send_message(unsigned msg,void* content)

{

do

{

int tmp = tail_pos;

MsgNode* pObj = &msg_buf[tmp];

if(pObj->flag != (unsigned)-1) //最后一个节点被使用了

{

//队列满

return -1;

}

}while(!__sync_bool_compare_and_swap(&tail_pos,tmp,(tmp + 1)%max_msg_size));

//申请一个消息记录,将消息记录赋值

pObj->content = content;

pObj->flag = msg; //此字段放最后赋值

return 0;

}

int CMessageQueue::get_message(MsgNode& msg)

{

do

{

int tmp = head_pos;

MsgNode* pObj = &msg_buf[tmp];

if(pObj->flag == (unsigned)-1)

{

//队列为空

return -1;

}

}while(!__sync_bool_compare_and_swap(&head_pos,tmp,(tmp + 1)%max_msg_size));

//重置消息

msg = *pObj;

pObj->Msg = (unsigned)-1;

return 0;

}

三、链表构成的队列：队列采用链表链式存储结构

注意：

1）此时的front和rear都变成了指针，front变成了头结点指针，而rear变成了尾节点的指针。

2）此处的front和rear类似于链表操作中的first和last。

3）入队实现主要在队列尾部实现，需要调整rear指针的指向；而出队操作主要在队头实现，需要调整front指针的指向。

判空： front==NULL;

判满： 如果不设定最大存储节点数，永远都为false;

入队： 要判断是否已满，同时要判断是否为第一个节点

出队： 要判断是否为空，同时要判断是否只有一个节点；nodeType * temp = front; Type value = front->info; front=front->link;if(temp==rear){rear==NULL;}delete temp;temp=NULL;

链表队列因为要同时修改front和rear指针，在首次插入时，没有想到比较好的办法。

链表队列的实现参考:

http://blog.163.com/xychenbaihu@yeah/blog/static/13222965520132343915442/

一写多读的无锁编程：

union CasStruct

{

unsigned int updateflag;

unsigned int updateindex;

CasStruct()

{

updateflag=0;

updateindex=0;

}

}

更新(写)：

开始更新：将updateflag设为1；结束更新：updateindex=(updateindex+1)%XXXXX；并将updateflag设为0。

读：

先获取一个CasHeader没有变化的对象，读完后比较是否发生变化，如果没有，说明读的过程中没有发生变化。