C#中使用CAS实现无锁算法的示例详解
CAS（Compare-and-Swap，比较并交换）是C#多线程并发编程中实现无锁同步的核心原子操作。本文将深入解析其原理，并展示如何利用System.Threading.Interlocked类构建线程安全的计数器与无锁并发队列，涵盖乐观锁策略下的自旋重试机制与数据一致性处理。
CAS 的基本概念与核心优势
CAS 是一种多线程常用的原子操作，用于在不使用显式互斥锁的情况下保证多线程间的同步与数据一致性。该操作通常包含三个关键参数：目标内存地址、期望的旧值以及准备写入的新值。逻辑定义为：
- CompareAndSwap(内存地址，期望的旧值，新值)
CAS 的运行机制如下：系统检查指定内存地址处的当前值是否等于期望的旧值。如果相等，则将该地址的值原子性地更新为新值；如果不相等，说明在此期间有其他线程修改了该数据，此时不执行写入操作，并通过返回值指示操作失败或返回实际旧值。这一比较和交换的过程由 CPU 硬件指令底层保证原子性，中间不会被其他线程中断。
相较于传统的锁机制（如 lock 关键字或 Mutex），CAS 具备显著优势：
- 无锁争用开销：无需额外的对象监视器，避免了加锁和解锁过程中的性能损耗及死锁风险。
- 无阻塞特性：当操作失败时，调用线程不会陷入操作系统级的阻塞与上下文切换，而是可通过自旋等待继续尝试，提高了高并发场景下的可伸缩性。
- 广泛适用性：可应用于自旋锁、计数器、队列等多种基础数据结构的设计中。
C# 中如何实现 CAS 操作
在 .NET 生态系统中，我们直接使用 System.Threading.Interlocked 静态类提供的原子方法来实现 CAS。该核心方法是 CompareExchange，它针对不同的数据类型（int, long, object 等）提供了重载版本。
// 以 int 类型为例
public static int CompareExchange(ref int location1, int value, int comparand);
// 以泛型引用类型为例
public static T CompareExchange<T>(ref T location1, T value, T comparand) where T : class;当调用 CompareExchange(ref location1, newValue, oldValue) 时：
- 系统读取
location1当前的值。 - 将其与传入的
comparand（即 old value）进行比较。 - 若两者相等，则将
value赋值给location1。 - 无论成功与否，方法都会立即返回
location1操作前的原始值。
程序员通过捕获方法的返回值并与初始传入的 comparand 进行比对，即可精确判断此次 CAS 操作是否成功完成。
无锁算法通用模板
单纯的 CAS 只能解决单一变量的原子更新。在实际业务中，通常需要“读取当前状态 -> 计算新状态 -> 尝试原子更新”的流程。由于连续的多步操作并非原子性（即步骤之间可能存在竞态条件），我们必须结合 while(true) 循环进行自旋重试：
- 目的：确保在任何时刻，数据的修改都建立在其未被其他线程篡改的前提上。
- 入口动作：每次进入循环，必须重新从内存中读取最新的状态变量。
- 核心验证：在计算新值后，发起 CAS 请求；并在必要时再次验证辅助状态（如队列指针）是否依然有效。
while (true)
{
// 1. 重新读取当前状态，因为前一次循环可能已被其他线程修改
var currentValue = GetCurrentValue();
// 2. 基于当前状态计算预期结果
var nextValue = CalculateNext(currentValue);
// 3. 尝试 CAS 替换
var actualResult = Interlocked.CompareExchange(ref targetLocation, nextValue, currentValue);
// 4. 只有当返回值等于我读取时的旧值时，才代表操作成功
if (actualResult == currentValue)
{
return nextValue;
}
}第1步：实现线程安全的无锁计数器
这是理解 CAS 最直观的入门案例。下面演示了一个简单的计数器类 Counter，其中使用了 volatile 修饰符以防止编译器重排序对读写的可见性问题，并结合 CAS 实现了纯粹的线程安全自增。
public class Counter
{
private volatile int _value;
public int Increment()
{
while (true)
{
// 始终读取最新的 _value
int oldValue = _value;
int newValue = oldValue + 1;
// 尝试将 newValue 换入。如果此时有人改了_value，CAS会失败
int result = Interlocked.CompareExchange(ref _value, newValue, oldValue);
// 成功标志：返回的实际旧值正是我们本轮开始时读取到的那个
if (result == oldValue)
{
return newValue;
}
}
}
}此模式广泛存在于 .NET Runtime 内部源码中。例如在 ThreadPool 管理线程池规模时，便依赖同样的逻辑来判断是否需要新增工作线程：
// 参考自 dotnet/runtime ThreadPoolWorkQueue.cs 的实现思想
internal void EnsureThreadRequested()
{
int count = ...; // 获取当前请求数
while (count < Environment.ProcessorCount)
{
// 尝试原子递增计数
int prev = Interlocked.CompareExchange(
ref _separated.numOutstandingThreadRequests,
count + 1,
count);
// 确认没有其他人在同一时刻抢占了这个计数的增加
if (prev == count)
{
ThreadPool.RequestWorkerThread();
break;
}
// 被抢占，更新本地 count 为最新值，下一轮重试
count = prev;
}
}第2步：实现基于链表的并发队列 (无锁 ConcurrentQueue)
在无锁并发队列的设计中，逻辑复杂度显著提升。我们采用经典的头尾伪节点设计：队首伪节点通过 _next 指向真实第一个元素，队尾伪节点用于标记末尾。这两个指针必须是 volatile 的。
public class ConcurrentQueue<T>
{
private volatile Node _head; // 队首标记
private volatile Node _tail; // 队尾标记
public ConcurrentQueue()
{
var dummy = new Node(default);
_head = _tail = dummy;
}
public void Enqueue(T item)
{
var node = new Node(item);
while (true)
{
Node tail = _tail;
Node next = tail._next;
// 验证点A：_tail是否被其他入队线程修改？
if (tail == _tail)
{
// 情况1：尾节点确实是最后一个，没有待挂载的后继节点
if (next == null)
{
// 尝试将我们的节点挂接到当前尾部的 _next
if (Interlocked.CompareExchange(ref tail._next, node, null) == null)
{
// 注意：这里即使失败了，只要没抛出异常就会回退到下一次循环重试
// 但为了加速，如果此时头部指针还是原来的 tail，说明推进成功
Interlocked.CompareExchange(ref _tail, node, tail);
break; // 成功入队
}
}
else
{
// 情况2：有其他人已经抢先一步更新了 next
// 尝试把 _tail 推向前去（这属于一种延迟清理的优化策略）
Interlocked.CompareExchange(ref _tail, next, tail);
}
}
}
}
public bool TryDequeue(out T item)
{
while (true)
{
Node head = _head;
Node tail = _tail;
Node next = head._next;
// 验证点B：_head是否被其他出队线程修改？
if (head == _head)
{
if (head == tail)
{
// 状态判定：空队列或者正在推进边界
if (next == null)
{
item = default;
return false; // 确实为空
}
// 辅助推进尾部指针
Interlocked.CompareExchange(ref _tail, next, tail);
}
else
{
// 取出节点值
item = next._item;
// 尝试弹出队首：用 next 替换掉 head
if (Interlocked.CompareExchange(ref _head, next, head) == head)
{
break; // 弹出成功
}
}
}
}
return true;
}
private class Node
{
public readonly T _item;
public Node _next;
public Node(T item) { _item = item; }
}
}对于上述实现的测试，可以使用多线程并行压力测试来验证数据完整性：
using System.Collections.Concurrent;
using System.Diagnostics;
var queue = new ConcurrentQueue<int>();
var results = new ConcurrentBag<int>();
int dequeueRetryCount = 0;
// 生产者任务
var enqueueTask = Task.Run(() =>
{
Thread.Sleep(10);
Parallel.For(0, 100000, i => queue.Enqueue(i));
});
// 消费者任务
var dequeueTask = Task.Run(() =>
{
Thread.Sleep(10);
Parallel.For(0, 100000, i =>
{
while (true)
{
if (queue.TryDequeue(out int result))
{
results.Add(result);
break;
}
Interlocked.Increment(ref dequeueRetryCount);
}
});
});
await Task.WhenAll(enqueueTask, dequeueTask);
Console.WriteLine($"Processed: {results.Count}, Retries: {dequeueRetryCount}");
// 校验唯一性
var hashSet = results.ToHashSet();
Debug.Assert(hashSet.Count == 100000);总结
CAS 操作本质上是一种乐观锁的思想体现：它假定数据通常不会被频繁竞争修改，因此在没有冲突时直接应用变更；一旦发生冲突，则放弃当前操作并从头开始重试。在借助 CAS 构建诸如队列、栈等复杂无锁数据结构时，不仅依赖于单一的原子交换，更需要精心安排每一步操作的时序，利用多次 if (pointer == loadedPointer) 式的状态快照校验，来抵御高并发环境下复杂的分支竞态问题。
以上就是C#中使用CAS实现无锁算法的示例详解的详细内容，更多关于C# CAS无锁算法的资料请关注本站其它相关文章！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
