在高并發場景下，傳統鎖機制帶來的線程阻塞和上下文切換開銷成為性能瓶頸。無鎖數據結構通過原子操作實現線程安全，避免了鎖的使用，成為高性能系統的關鍵技術。本文將深入探討C++中基于CAS（Compare-and-Swap）的無鎖數據結構實現原理、應用場景及實戰技巧。
一、CAS原理解析
1.1 什么是CAS？
CAS是一種原子操作，包含三個參數：內存地址(V)、舊值(A)和新值(B)。操作邏輯如下：
- 如果內存地址V中的值等于舊值A，則將該位置的值更新為新值B
- 否則返回當前內存中的實際值
CAS操作通常由CPU硬件直接支持，是無鎖編程的核心原語。
1.2 C++中的CAS實現
C++11在<atomic>頭文件中提供了原子操作支持，其中compare_exchange_weak和compare_exchange_strong對應CAS操作：
#include <atomic>template<typename T>
bool atomic_compare_and_swap(std::atomic<T>& atom, T& expected, T desired) {return atom.compare_exchange_weak(expected, desired);
}
compare_exchange_weak：可能會虛假失敗（即使值相等也可能返回false），但性能更高compare_exchange_strong：保證不會虛假失敗，但可能需要更多次重試
1.3 CAS的ABA問題
CAS操作存在ABA問題：當值從A變為B再變回A時，CAS會誤認為值沒有變化。解決方法通常是使用帶版本號的原子變量：
template<typename T>
struct VersionedValue {T value;uint64_t version;bool operator==(const VersionedValue& other) const {return value == other.value && version == other.version;}
};std::atomic<VersionedValue<int>> atom;// 更新操作
void update(int new_value) {VersionedValue<int> expected = atom.load();VersionedValue<int> desired;do {desired.value = new_value;desired.version = expected.version + 1;} while (!atom.compare_exchange_weak(expected, desired));
}
二、基于CAS的無鎖棧實現
2.1 單鏈表節點結構
template<typename T>
struct Node {T data;Node* next;Node(const T& value) : data(value), next(nullptr) {}
};
2.2 無鎖棧核心實現
template<typename T>
class LockFreeStack {
private:std::atomic<Node<T>*> head;public:LockFreeStack() : head(nullptr) {}// 入棧操作void push(const T& value) {Node<T>* new_node = new Node<T>(value);new_node->next = head.load();while (!head.compare_exchange_weak(new_node->next, new_node)) {// CAS失敗，說明有其他線程修改了head// 重新加載head并嘗試}}// 出棧操作bool pop(T& value) {Node<T>* old_head = head.load();while (old_head != nullptr) {// 嘗試將head指向下一個節點if (head.compare_exchange_weak(old_head, old_head->next)) {value = old_head->data;delete old_head;return true;}}return false; // 棧為空}
};
2.3 ABA問題處理
上述實現存在ABA問題，改進版使用帶標記的指針：
template<typename T>
class LockFreeStack {
private:struct TaggedPointer {Node<T>* ptr;uint64_t tag;bool operator==(const TaggedPointer& other) const {return ptr == other.ptr && tag == other.tag;}};std::atomic<TaggedPointer> head;public:void push(const T& value) {Node<T>* new_node = new Node<T>(value);TaggedPointer old_head = head.load();do {new_node->next = old_head.ptr;TaggedPointer new_head = {new_node, old_head.tag + 1};} while (!head.compare_exchange_weak(old_head, new_head));}bool pop(T& value) {TaggedPointer old_head = head.load();while (old_head.ptr != nullptr) {TaggedPointer new_head = {old_head.ptr->next, old_head.tag + 1};if (head.compare_exchange_weak(old_head, new_head)) {value = old_head.ptr->data;delete old_head.ptr;return true;}}return false;}
};
三、無鎖隊列實現
3.1 Michael-Scott無鎖隊列
template<typename T>
class LockFreeQueue {
private:struct Node {T data;std::atomic<Node*> next;Node() : next(nullptr) {}Node(const T& value) : data(value), next(nullptr) {}};std::atomic<Node*> head;std::atomic<Node*> tail;public:LockFreeQueue() {Node* dummy = new Node();head.store(dummy);tail.store(dummy);}~LockFreeQueue() {T value;while (pop(value));delete head.load();}// 入隊操作void enqueue(const T& value) {Node* new_node = new Node(value);Node* old_tail = tail.load();while (true) {// 嘗試將新節點連接到隊尾if (old_tail->next.compare_exchange_weak(nullptr, new_node)) {// 成功連接，嘗試更新tail指針tail.compare_exchange_strong(old_tail, new_node);return;} else {// 連接失敗，說明有其他線程已經更新了tailtail.compare_exchange_strong(old_tail, old_tail->next);old_tail = tail.load();}}}// 出隊操作bool dequeue(T& value) {Node* old_head = head.load();while (true) {Node* next = old_head->next.load();if (next == nullptr) {return false; // 隊列為空}// 嘗試更新head指針if (head.compare_exchange_weak(old_head, next)) {value = next->data;delete old_head; // 釋放原頭節點（dummy或前一個節點）return true;}}}
};
四、CAS無鎖算法的性能考量
4.1 優勢
- 無阻塞：線程不會因競爭而阻塞，減少上下文切換開銷
- 細粒度并發：允許多個線程同時操作不同部分的數據結構
- 避免死鎖：無需獲取鎖，從根本上消除死鎖風險
4.2 劣勢
- 高競爭下性能下降：大量CAS失敗導致頻繁重試
- 實現復雜度高：需要處理ABA問題、內存回收等復雜問題
- 調試困難：非確定性的執行順序增加調試難度
4.3 適用場景
- 高并發低競爭：線程間沖突較少的場景
- 實時系統：不允許長時間阻塞的場景
- 高性能核心組件：如網絡框架、數據庫引擎
五、內存回收方案
無鎖數據結構的一個關鍵挑戰是安全地回收內存，避免"懸垂指針"問題。常見方案有：
5.1 Hazard Pointer
// 簡化版Hazard Pointer實現
template<typename T>
class HazardPointer {
private:static constexpr int MAX_THREADS = 64;std::atomic<void*> hazard_pointers[MAX_THREADS];std::atomic<int> thread_count;thread_local static int thread_id;public:void* get_hazard_pointer() {if (thread_id == -1) {thread_id = thread_count.fetch_add(1);}return hazard_pointers[thread_id].load();}void set_hazard_pointer(void* ptr) {hazard_pointers[thread_id].store(ptr);}void clear_hazard_pointer() {hazard_pointers[thread_id].store(nullptr);}bool can_reclaim(Node<T>* node) {for (int i = 0; i < thread_count; ++i) {if (hazard_pointers[i].load() == node) {return false;}}return true;}
};
5.2 延遲刪除
// 簡化版延遲刪除實現
template<typename T>
class LockFreeStack {
private:std::atomic<Node<T>*> head;std::atomic<int> delete_count;std::vector<Node<T>*> to_be_deleted;const int BATCH_SIZE = 100;public:bool pop(T& value) {Node<T>* old_head = head.load();while (old_head != nullptr) {if (head.compare_exchange_weak(old_head, old_head->next)) {value = old_head->data;// 延遲刪除to_be_deleted.push_back(old_head);if (++delete_count >= BATCH_SIZE) {perform_deletion();}return true;}}return false;}void perform_deletion() {std::vector<Node<T>*> local;local.swap(to_be_deleted);delete_count = 0;for (auto node : local) {delete node;}}
};
六、總結與最佳實踐
- 優先使用標準庫實現：C++標準庫提供了許多無鎖容器，如
std::atomic、std::queue的并發版本 - 謹慎使用無鎖結構：僅在性能關鍵且鎖競爭激烈的場景使用
- 解決ABA問題：使用帶版本號的指針或Hazard Pointer
- 注意內存回收：避免使用裸指針，采用安全的內存回收機制
- 充分測試：無鎖算法的正確性依賴于原子操作的順序，需進行充分測試
無鎖數據結構是一把雙刃劍，正確使用能帶來顯著性能提升，但實現難度較大。開發者需根據具體場景權衡利弊，選擇合適的實現方案。隨著硬件和編程語言的發展，無鎖編程將在高性能領域發揮越來越重要的作用。
