简单的无锁队列环形实现

/*
 * RingBuf.h
 *
 *  Created on: Feb 7, 2015 6:06:10 PM
 *      Author: xuzewen
 */

#ifndef RINGBUF_H_
#define RINGBUF_H_

#include <libmisc.h>

/**
 *
 * 多/单线程生产, 仅仅能单线程消费, 尺寸固定为0x10000.
 *
 * */
class RingBuf
{
public:
	ushort ccur; /** 消费游标. */
	uint pcur; /** 生产游标. */
	size_t* ring;
public:
	RingBuf();
	virtual ~RingBuf();
public:
	bool push(void* e); /** 加入, 重复尝试, 直到成功为止. */
	void* pop(); /** 弾出, 总是马上返回, 无元素时返回空. */
	int size(); /** 返回元素个数. */
};

#endif /* RINGBUF_H_ */

实现:

/*
 * RingBuf.cpp
 *
 *  Created on: Feb 7, 2015 6:06:10 PM
 *      Author: xuzewen
 */

#include "RingBuf.h"

RingBuf::RingBuf()
{
	this->pcur = 0;
	this->ccur = 0;
	this->ring = (size_t*) calloc(1, sizeof(size_t) * 0x10000);
}

/** 加入, 重复尝试, 直到成功为止. */
bool RingBuf::push(void* f)
{
	while (!__sync_bool_compare_and_swap(this->ring + (this->pcur & 0x0000FFFF), 0, (size_t) f))
		usleep(1000);
	//__sync_fetch_and_add(&this->pcur, 1);	/** 多线程生产者. */
	//
	++this->pcur; /** 单线程的生产者. */
	return true;
}

/** 弾出, 总是马上返回, 无元素时返回空. */
void* RingBuf::pop()
{
	if (this->ring[this->ccur] == 0)
		return NULL;
	void* r = (void*) this->ring[this->ccur];
	this->ring[this->ccur] = 0;
	++this->ccur;
	return r;
}

int RingBuf::size()
{
	int size = 0;
	for (int i = 0; i < 0x10000; ++i)
	{
		if (this->ring[this->ccur] != 0)
			++size;
	}
	return size;
}

RingBuf::~RingBuf()
{

}

測试:

/*
 * main.cpp
 *
 *  Created on: Mar 20, 2015 12:09:33 PM
 *      Author: xuzewen
 */

#include "RingBuf.h"

static RingBuf* r = new RingBuf();

void svc()
{
	ullong last = 0;
	ullong ts = 0;
	while (1)
	{
		ullong c = (ullong) r->pop();
		if (c == 0) /** not NULL. */
		{
			usleep(100);
			continue;
		}
		if (c - last != 1)
		{
			printf("it`s a bug, c: %llu, last: %llu
", c, last);
			exit(1);
		}
		last = c;
		if ((c % 20000000) == 0)
		{
			ullong now = Misc::getDiDa(); /** 系统嘀嗒声. */
			printf("ts: %llu, last: %llu
", now - ts, last);
			ts = now;
		}
	}
}

int main(int argc, char **argv)
{
	Misc::newThread(svc); /** 单线程消费. */
	//
	for (ullong c = 1;; ++c) /** 单线程生产. */
		r->push((void*) c); /** 按顺序push. */
	return EXIT_SUCCESS;
}

測试结果:

按每2000万条消息, 消耗470毫秒算, 并发数为4255万条/秒. 单线程生产, 单线程消费, 两个线程接近跑满.