首页 > 代码库 > 使用无锁队列（环形缓冲区）注意事项
使用无锁队列（环形缓冲区）注意事项
环形缓冲区是生产者和消费者模型中常用的数据结构。生产者将数据放入数组的尾端，而消费者从数组的另一端移走数据，当达到数组的尾部时，生产者绕回到数组的头部。如果只有一个生产者和一个消费者，那么就可以做到免锁访问环形缓冲区（Ring Buffer）。写入索引只允许生产者访问并修改，只要写入者在更新索引之前将新的值保存到缓冲区中，则读者将始终看到一致的数据结构。同理，读取索引也只允许消费者访问并修改。
环形缓冲区实现原理图
如图所示，当读者和写者指针相等时，表明缓冲区是空的，而只要写入指针在读取指针后面时，表明缓冲区已满。
清单 9. 2.6.10 环形缓冲区实现代码
/*
* __kfifo_put - puts some data into the FIFO, no locking version
* Note that with only one concurrent reader and one concurrent
* writer, you don‘t need extra locking to use these functions.
*/
unsigned int __kfifo_put(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->size - fifo->in + fifo->out);
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l);
fifo->in += len;
return len;
}
/*
* __kfifo_get - gets some data from the FIFO, no locking version
* Note that with only one concurrent reader and one concurrent
* writer, you don‘t need extra locking to use these functions.
*/
unsigned int __kfifo_get(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->in - fifo->out);
/* first get the data from fifo->out until the end of the buffer */
l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
memcpy(buffer, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
/* then get the rest (if any) from the beginning of the buffer */
memcpy(buffer + l, fifo->buffer, len - l);
fifo->out += len;
return len;
}
需要注意的是
使用ring_buffer_get（kfifo_get）或者ring_buffer_put（kfifo_put）时，如果返回参数与传入参数len不相等时，则操作失败
我们定义一个
//注意student_info 共17字节 按照内存排列占24字节
typedef struct student_info
{
uint64_t stu_id; //8个字节
uint32_t age; //4字节
uint32_t score;//4字节
char sex;//1字节
}student_info;
我们建立一个环形缓冲区，里面只有64字节大小（虽然我们实际使用时大小远大于此），向里面多次存入24字节student_info，看有什么反应
//打印学生信息
void print_student_info(const student_info *stu_info)
{
assert(stu_info);
printf("id:%lu\t",stu_info->stu_id);
printf("age:%u\t",stu_info->age);
printf("sex:%d\t",stu_info->sex);
printf("score:%u\n",stu_info->score);
}
student_info * get_student_info(time_t timer)
{
student_info *stu_info = (student_info *)malloc(sizeof(student_info));
srand(timer);
stu_info->stu_id = 10000 + rand() % 9999;
stu_info->age = rand() % 30;
stu_info->score = rand() % 101;
stu_info->sex=rand() % 2;
print_student_info(stu_info);
return stu_info;
}
void print_ring_buffer_len(struct ring_buffer *ring_buf)
{
//用于打印缓冲区长度
uint32_t ring_buf_len = 0;
//取得已经使用缓冲区长度 size-ring_buf_len为未使用缓冲区的长度
ring_buf_len=ring_buffer_len(ring_buf);
printf("no use ring_buf_len:%d\n",(ring_buf->size-ring_buf_len));
}
int main(int argc, char *argv[])
{
uint32_t size = 0;
//用于判断存储或者取得数据的字节数
uint32_t oklen = 0;
struct ring_buffer *ring_buf = NULL;
//64字节
size=BUFFER_SIZE;
ring_buf = ring_buffer_alloc(size);
printf("input student\n");
{
student_info *stu_info;
student_info stu_temp;
uint32_t student_len=sizeof(student_info);
printf("ring_buf_len:%d\n",ring_buf->size);
printf("student_len:%d\n",student_len);
//此时环形缓冲区没有数据我们去取数据当然为空
memset(&stu_temp,0,student_len);
oklen=ring_buffer_get(ring_buf, (void *)(&stu_temp), student_len);
if(oklen==student_len)
{
printf("get student data\n");
}
else
{
printf("no student data\n");
}
printf("\n");
//第一次调用时用字节结束后还有64-24 =40字节
stu_info = get_student_info(976686458);
oklen = ring_buffer_put(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
printf("1 put student data success\n");
}
else
{
printf("1 put student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
//第二次调用时用字节结束后还有64-48 =16字节
stu_info = get_student_info(976686464);
oklen= ring_buffer_put(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
printf("2 put student data success\n");
}
else
{
printf("2 put student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
//第三次调用时需要用字节但只有字节失败
//把字节都写满了
//验证了在调用__kfifo_put函数或者__kfifo_get函数时，如果返回参数与传入参数len不相等时，则操作失败
stu_info = get_student_info(976686445);
oklen= ring_buffer_put(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
printf("3 put student data success\n");
}
else
{
printf("3 put student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
//第四次调用时需要用字节但无字节
////验证了在调用__kfifo_put函数或者__kfifo_get函数时，如果返回参数与传入参数len不相等时，则操作失败
stu_info = get_student_info(976686421);
oklen= ring_buffer_put(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
printf("4 put student data success\n");
}
else
{
printf("4 put student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
//现在开始取学生数据里面保存了个学生数据我们取三次看效果
printf("output student\n");
printf("\n");
//第一次取得数据并打印
memset(stu_info,0,student_len);
oklen=ring_buffer_get(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
print_student_info(stu_info);
printf("1 get student data success\n");
}
else
{
printf("1 get student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
////第二次取得数据并打印
memset(stu_info,0,student_len);
oklen=ring_buffer_get(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
print_student_info(stu_info);
printf("2 get student data success\n");
}
else
{
printf("2 get student data failure\n");
}
print_ring_buffer_len(ring_buf);
printf("\n");
//第三次取得数据失败
memset(stu_info,0,student_len);
oklen=ring_buffer_get(ring_buf, (void *)stu_info, student_len);
if(oklen==student_len)
{
print_student_info(stu_info);
printf("3 get student data success\n");
}
else
{
printf("3 get student data failure\n");
}
print_ring_buffer_len(ring_buf);
}
return 1;
}
结论：在使用ring_buffer_get（kfifo_get）或者ring_buffer_put（kfifo_put）时，如果返回参数与传入参数len不相等时，则操作失败。代码下载：tessc.rar（http://files.cnblogs.com/dragonsuc/TDDOWNLOAD.zip）
需要注意的地方：
1.只有一个线程负责读，另一个线程负责写的时候，数据是线程安全的。上面的实现是基于这个原理实现的，当有多个线程读或者多个线程写的时候，不保证数据的正确性。
所以使用的时候，一个线程写，一个线程读。网络应用中比较常用，就是开一个线程接口数据，然后把数据写入队列。然后开一个调度线程读取网络数据，然后分发到处理线程。
2.数据长度默认宏定义了一个长度，超过这个长度的时候，后续的数据会写入失败。
本文参考文章：
http://blog.csdn.net/mergerly/article/details/39009473
http://www.cnblogs.com/Anker/p/3481373.html
使用无锁队列（环形缓冲区）注意事项