无锁编程案例

本文目录导读：

1. 无锁队列（Lock-Free Queue）
2. 无锁栈（Lock-Free Stack）
3. 无锁计数器（Lock-Free Counter）
4. 无锁HashMap（简化版）
5. 无锁读写计数器（用于缓存）
6. 使用示例
7. 关键要点

我将为您展示几个经典的无锁编程案例,涵盖不同的数据结构和应用场景。

无锁队列（Lock-Free Queue）

Michael-Scott 队列实现

#include <atomic>
#include <memory>
#include <optional>
template<typename T>
class LockFreeQueue {
private:
    struct Node {
        std::shared_ptr<T> data;
        std::atomic<Node*> next;
        Node() : next(nullptr) {}
        Node(const T& value) : data(std::make_shared<T>(value)), next(nullptr) {}
    };
    std::atomic<Node*> head;
    std::atomic<Node*> tail;
public:
    LockFreeQueue() {
        Node* dummy = new Node();
        head.store(dummy);
        tail.store(dummy);
    }
    ~LockFreeQueue() {
        while (Node* old_head = head.load()) {
            head.store(old_head->next);
            delete old_head;
        }
    }
    void enqueue(const T& value) {
        Node* new_node = new Node(value);
        new_node->next.store(nullptr);
        while (true) {
            Node* last = tail.load();
            Node* next = last->next.load();
            // 确认tail仍然指向最后一个节点
            if (last == tail.load()) {
                if (next == nullptr) {
                    // 尝试将新节点链接到尾部
                    if (last->next.compare_exchange_weak(next, new_node)) {
                        // 成功链接，尝试移动tail
                        tail.compare_exchange_weak(last, new_node);
                        return;
                    }
                } else {
                    // tail落后，尝试推进tail
                    tail.compare_exchange_weak(last, next);
                }
            }
        }
    }
    std::optional<T> dequeue() {
        while (true) {
            Node* first = head.load();
            Node* last = tail.load();
            Node* next = first->next.load();
            // 检查head是否仍然是有效的
            if (first == head.load()) {
                if (first == last) {
                    // 队列可能为空
                    if (next == nullptr) {
                        return std::nullopt;
                    }
                    // tail落后于head
                    tail.compare_exchange_weak(last, next);
                } else {
                    T value = *next->data;
                    // 尝试移动head
                    if (head.compare_exchange_weak(first, next)) {
                        // 安全地删除节点（在实际生产环境中需要考虑内存回收）
                        delete first;
                        return value;
                    }
                }
            }
        }
    }
};

无锁栈（Lock-Free Stack）

Treiber Stack 实现

#include <atomic>
#include <memory>
template<typename T>
class LockFreeStack {
private:
    struct Node {
        T value;
        std::atomic<Node*> next;
        Node(const T& val) : value(val), next(nullptr) {}
    };
    std::atomic<Node*> top;
    // 用于内存回收（Harzard Pointer的简化版本）
    std::atomic<Node*> retired_list;
public:
    LockFreeStack() : top(nullptr), retired_list(nullptr) {}
    void push(const T& value) {
        Node* new_node = new Node(value);
        new_node->next.store(top.load());
        // CAS循环
        while (!top.compare_exchange_weak(new_node->next.load(), new_node)) {
            // 如果CAS失败，new_node->next会被更新为最新的top值
        }
    }
    std::optional<T> pop() {
        while (true) {
            Node* current_top = top.load();
            if (current_top == nullptr) {
                return std::nullopt; // 栈为空
            }
            T value = current_top->value;
            Node* next = current_top->next.load();
            // 尝试更新top指针
            if (top.compare_exchange_weak(current_top, next)) {
                // 内存回收（在实际应用中需要考虑线程安全问题）
                delete current_top;
                return value;
            }
            // 如果CAS失败，重试
        }
    }
};

无锁计数器（Lock-Free Counter）

使用Fetch-Add实现

#include <atomic>
#include <thread>
class AtomicCounter {
private:
    std::atomic<long long> count{0};
public:
    // 原子递增（fetch-add 操作）
    long long increment() {
        return count.fetch_add(1, std::memory_order_relaxed);
    }
    // 原子递减
    long long decrement() {
        return count.fetch_sub(1, std::memory_order_relaxed);
    }
    // 读取当前值
    long long get() const {
        return count.load(std::memory_order_acquire);
    }
    // 重置计数
    void reset() {
        count.store(0, std::memory_order_release);
    }
};
// 无锁计数器示例：统计请求次数
class RequestCounter {
private:
    std::atomic<int> active_requests{0};
    std::atomic<int> total_requests{0};
public:
    void request_start() {
        active_requests.fetch_add(1, std::memory_order_relaxed);
        total_requests.fetch_add(1, std::memory_order_relaxed);
    }
    void request_end() {
        active_requests.fetch_sub(1, std::memory_order_relaxed);
    }
    int get_total() const {
        return total_requests.load(std::memory_order_relaxed);
    }
    bool is_idle() const {
        return active_requests.load() == 0;
    }
};

无锁HashMap（简化版）

#include <atomic>
#include <vector>
template<typename K, typename V>
class LockFreeHashMap {
private:
    struct Node {
        K key;
        V value;
        std::atomic<Node*> next;
        Node(const K& k, const V& v) : key(k), value(v), next(nullptr) {}
    };
    std::vector<std::atomic<Node*>> buckets;
    size_t capacity;
    size_t hash(const K& key) const {
        return std::hash<K>{}(key) % capacity;
    }
public:
    LockFreeHashMap(size_t size = 16) : capacity(size), buckets(size) {
        for (auto& bucket : buckets) {
            bucket.store(nullptr);
        }
    }
    void insert(const K& key, const V& value) {
        size_t idx = hash(key);
        Node* new_node = new Node(key, value);
        while (true) {
            Node* current_head = buckets[idx].load();
            new_node->next.store(current_head);
            // 检查键是否已存在
            Node* probe = current_head;
            while (probe && probe->key != key) {
                probe = probe->next.load();
            }
            if (probe) {
                // 键已存在，可以更新值（这里简化为不更新）
                delete new_node;
                return;
            }
            // CAS插入
            if (buckets[idx].compare_exchange_weak(current_head, new_node)) {
                return;
            }
        }
    }
    std::optional<V> find(const K& key) const {
        size_t idx = hash(key);
        Node* current = buckets[idx].load();
        while (current) {
            if (current->key == key) {
                return current->value;
            }
            current = current->next.load();
        }
        return std::nullopt;
    }
};

无锁读写计数器（用于缓存）

#include <atomic>
#include <chrono>
#include <thread>
class CacheEntry {
private:
    std::atomic<long long> access_count;
    std::atomic<long long> miss_count;
    std::atomic<bool> is_cached;
public:
    CacheEntry() : access_count(0), miss_count(0), is_cached(false) {}
    void record_access() {
        access_count.fetch_add(1, std::memory_order_relaxed);
    }
    void record_miss() {
        miss_count.fetch_add(1, std::memory_order_relaxed);
    }
    void set_cached() {
        is_cached.store(true, std::memory_order_release);
    }
    double get_hit_rate() const {
        long long accesses = access_count.load(std::memory_order_acquire);
        long long misses = miss_count.load(std::memory_order_acquire);
        if (accesses == 0) return 0.0;
        return static_cast<double>(accesses - misses) / accesses * 100.0;
    }
    void reset() {
        access_count.store(0, std::memory_order_release);
        miss_count.store(0, std::memory_order_release);
    }
};

使用示例

int main() {
    // 测试无锁队列
    LockFreeQueue<int> queue;
    // 生产者
    std::thread producer([&]() {
        for (int i = 0; i < 1000; ++i) {
            queue.enqueue(i);
        }
    });
    // 消费者
    std::thread consumer([&]() {
        int count = 0;
        while (count < 1000) {
            auto value = queue.dequeue();
            if (value) {
                count++;
            } else {
                std::this_thread::yield();
            }
        }
    });
    producer.join();
    consumer.join();
    // 测试无锁计数器
    AtomicCounter counter;
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back([&]() {
            for (int j = 0; j < 100; ++j) {
                counter.increment();
            }
        });
    }
    for (auto& t : threads) {
        t.join();
    }
    std::cout << "Final count: " << counter.get() << std::endl;
    return 0;
}

关键要点

1. CAS（Compare-And-Swap）：所有无锁编程的基础操作
2. 内存序：正确使用memory_order很重要（relaxed、acquire、release）
3. ABA问题：需要特别注意内存回收
4. 内存回收：无锁结构的内存管理是最大挑战
5. 性能考量：在高竞争环境下，无锁结构可能并不比锁机制快

这些案例展示了无锁编程的主要应用场景,但在实际生产环境中，还需要考虑更多细节，如ABA问题解决、内存回收机制等。