最近想在DIOCP中加入任务调度线程，DIOCP的工作线程作为生产者(producer)将接受到的数据对象，投递到任务调度线程中，然后统一进行分配。然而这一切都需要一个队列， 这几天都在关注无锁队列。
[队列]
首先是一个队列，简单的队列就是，生产者把数据压入队列(push), 消费者通过队列Pop出数据进行处理。
简单的队列就是提供Push，和Pop函数。
我们用一个链表来存储数据。Head ->data01->data02...data_n->Tail, 每个数据块的结构如下
type
PVarQueueBlock = ^TVarQueueBlock;
TVarQueueBlock = packed record
value:Variant;
next:PVarQueueBlock;
end;
1.在进行Push压入数据时压入将Tail.next指向新压入的数据块， 然后用新的数据块做Tail
procedure TSimpleQueue.pushQueue(pvData: PVarQueueBlock);
begin
if FHead = nil then
begin
FHead = pvData;
FTail := FHead;
end else
begin
FTail.next := pvData;
FTail := pvData;
end;
Inc(FCount);
end;
2.在进行Pop数据时把Head数据块取出，然后用Head数据块指向的下一块当作Head.
function TSimpleQueue.popQueue: PVarQueueBlock;
var
lvTemp, lvRet:PVarQueueBlock;
begin
lvTemp := FHead;
if (lvTemp = nil) then
begin //没有任何可以Pop出的值
Result := nil;
exit;
end; //
FHead := FHead.next; Dec(FCount);
Result :=lvTemp;
end;
上面就是简单的队列
[无锁队列]
上面的实现的队列在多线程情况下是不安全的。如果要在多并发下队列要进行加锁，在push和pop时加锁也是一种办法。可以直接用临界就可以了，但是我们要做的是无锁队列
首先记住多并发设计规则：决不要假设任何代码会连续执行
上面的push操作
FTail.next := pvData;
FTail := pvData;
也许执行了FTail.next:=pvData后，会被另外的线程抢走，然后FTail进行了新的赋值，这样在进行FTail := pvData;这样整个数据链条就会被破坏。
如果这两行我们能一次行完成，这样就可以实现无锁操作了，这样我们需要引入原子操作.Interlocked中的函数。
说无锁其实不太确切，只是锁的粒度小了。我们是使用api的InterlockedCompareExchange函数来实现的。
查一下MSDN
http://msdn.microsoft.com/en-us/library/windows/desktop/ms683560(v=vs.85).aspx
LONG __cdecl InterlockedCompareExchange(
_Inout_ LONG volatile *Destination,
_In_ LONG Exchange,
_In_ LONG Comparand
);
Parameters
- Destination [in, out]
-
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.
大概解释一下。这个函数是比较后进行交换。第一个参数是要存放目的的数据，第二个是交换数据，第三个是比较数据(与第一个比较), 如果交换返回的数据和第三个参数一样。
这样就可以在push和pop一步完成。
这里贴出push的pop操作
procedure TVarQueue.pushQueue(pvData: PVarQueueBlock);
var
lvTemp:PVarQueueBlock;
lvPointer:Pointer;
begin
while True do
begin
lvTemp := FTail;
while lvTemp.next <> nil do lvTemp := lvTemp.next;
if InterlockedCompareExchangePointer(Pointer(lvTemp.next), Pointer(pvData), nil) = nil then
begin
break;
end;
end;
FTail := pvData;
Inc(FCount);
end;
function TVarQueue.popQueue: PVarQueueBlock;
var
lvTemp, lvRet:PVarQueueBlock;
lvPointer:Pointer;
begin
///为了方便 队列中始终保留一个FHead数据块
/// 也就是说FHead指向的下一个数据块才是第一个数据块
/// while True do
begin
lvTemp := FHead;
if (lvTemp = nil) or (lvTemp.next = nil) then
begin //没有任何可以Pop出的值
Result := nil;
exit;
end;
if InterlockedCompareExchangePointer(Pointer(FHead), lvTemp.next, lvTemp) = lvTemp then
begin
break;
end;
end;
Dec(FCount);
lvRet := lvTemp.next;
Result := lvRet;
lvTemp.next := nil;
Dispose(lvTemp);
//返回的是head.next
end;
后续我会上传完整的代码到DIOCP项目中。
如有漏洞，敬请指出。欢迎假如DIOCP群讨论