无锁队列原理及实现（三）

第二篇讲了环形队列，也就是环形缓冲区RingBuffer的实现。结尾提到了，如果仅仅是数据结构上并不能保证在多线程情况下的线程安全。需要加入同步机制。那么这一篇主要就要来讨论一下何如正确的保证线程安全的使用无所队列。

使用队列最简单的方式就是想到在操作队列时进行加锁，也就是在进行插入或者取出操作时先要获得锁，才能够对队列进行操作，这样就能保证里面的数据在同一时刻只能被一个线程所改变。

下面的代码分别代表了一个producer和一个consumer线程。其使用互斥锁进行同步的方式。

 
void ProducerRoutine(ThreadContext* pContext)
{	
    ...
 
    pContext->pMtx->lock();
	if (!pContext->pRingBuffer->isFull())
	{
		pContext->pRingBuffer->insert(pDataBuffer[count]);
	}	
	pContext->pMtx->unlock();
 
	...
}
 
void ComsumerRoutine(ThreadContext* pContext)
{
    ...
 
    pContext->pMtx->lock();
	if (!pContext->pRingBuffer->isEmpty())
	{
    	pContext->pRingBuffer->fetch(pDataBuffer[count]);
	}
	pContext->pMtx->unlock();
 
	...
}

当然这种方式肯定时可行，并且有效的。而且在不少代码里面，都会看到类似的操作。但是锁机制会使用内核对象进行同步开销比较大。

因此就引入了CAS比较并替换的方式来实现线程安全的操作。

CAS的核心原理就是对指定的内存进行比较并替换的原子操作。主流的处理器均有指令集上的实现。因此不需要进行内核操作。在Windows平台上提供了以_Interlocked开头的一系列原子操作函数，这里我们用到的函数为_InterlockedCompareExchange，其定义如下

// 如果Destination指向的值和Comparand的值相等，则用Exchange的值去替换它。
// 返回值 Destination的初始值
char _InterlockedCompareExchange8(
   char volatile * Destination,    // 需要对比并交换的目标地址
   char Exchange,                  // 需要交换的值
   char Comparand                  // 所对比的值
);

有了这个InterlockedCompareExchange这个原子操作我们就可以将同步锁，例如Mutex替换为CAS操作。编译之后为cmpxchg，例如

lock cmpxchg byte ptr [rcx+1Ch],r14b

如果简单的我们只是把该CAS作为一个锁变量，则可以这样实现。

bool RingBuffer::getLock()
{
	if (_InterlockedCompareExchange8(&m_bGuard, true, false)) 
	{
		return false;
	}
 
	return m_bGuard;
}
 
void RingBuffer::releaseLock()
{
	_InterlockedCompareExchange8(&m_bGuard, false, true);
}

即简单的创建一个m_bGuard成员作为标志，如果该标志被占用，则不允许对buffer进行操作。而加这个Guard的地方也和锁一致。也是需要使用时获取，然后使用完释放。

虽然这样能够提升一定的效率，但是还是没有解决实际的问题。基本上同时也只会有一个线程对RingBuffer进行操作。我们想要达到的效果是同一时间能够有两个线程对Ring进行操作。

让我们再回顾一下之前我们实现的读线程和写线程分别的操作。

读线程

1. 是否空
2. 移动指针
3. 读取数据

写线程

1. 是否满
2. 移动指针
3. 读取线程

其实会发现，改动指针值只会发生在固定的线程。而就算值没有来得及刷新修改的值，也只会让另外一个线程没有办法及时操作。不会出现修改同一个值的情况。不过如果只是简简单单的直接让两个线程来操作，还是会出现结果不正常的情况。原因是在于之前我们实现的方式。当程序处于初始状态时，看看下面流程

1. 写线程：不为满
2. 写线程：移动
3. 读线程：不为空
4. 读线程：移动
5. 读线程：读取
6. 写线程：写入

自然最终读线程读取到的内容实际上时还没有新写入的内容。因此操作内容必须在移动指针之前进行。这样就不会存在多线程打断操作的情况，移动完毕，代表真的操作完成。

最后，需要注意的是，因为判断和写入都是作为判断的依据，因此需要将指针的数据设定为volatile，不会让编译器进行优化，保证能够及时写入内存里面，让另一个线程能够进行判断时读取到的是及时修改的指针内容。连前面的guard都不需要了。