C++ 减少锁竞争：无锁数据结构与原子操作的优先级

C++ 减少锁竞争：无锁数据结构与原子操作的优先级 (讲座版)

大家好！今天咱们来聊聊C++里让人头疼又不得不面对的锁竞争问题，以及如何通过无锁数据结构和原子操作来优雅地解决它。

第一幕：锁，爱恨交织的家伙

在多线程的世界里，共享资源就像一块大蛋糕，谁都想咬一口。但是，如果大家都扑上去抢，蛋糕肯定会被弄得乱七八糟。为了保证蛋糕的完整性，我们引入了锁。

锁的作用很简单，就是让同一时间只有一个线程能访问共享资源。这就像给蛋糕加上了一把锁，只有拿到钥匙（锁）的线程才能享用。

#include <iostream>
#include <thread>
#include <mutex>

std::mutex mtx; // 一把锁

int shared_data = 0;

void increment() {
    for (int i = 0; i < 100000; ++i) {
        mtx.lock(); // 获取锁
        shared_data++;
        mtx.unlock(); // 释放锁
    }
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);

    t1.join();
    t2.join();

    std::cout << "Shared data: " << shared_data << std::endl; // 预期输出: 200000
    return 0;
}

这段代码展示了最基本的锁的使用。两个线程同时尝试增加shared_data，通过锁来保证数据的正确性。

但是，锁这玩意儿，用好了是守护神，用不好就是绊脚石。锁竞争会导致线程阻塞，白白浪费CPU时间，降低程序的性能。就好比大家都排队等着拿蛋糕，队伍越长，大家等待的时间就越长。

第二幕：锁竞争，性能的隐形杀手

锁竞争究竟是怎么影响性能的呢？主要有以下几个方面：

上下文切换： 当一个线程尝试获取一个已经被其他线程持有的锁时，它会被阻塞，操作系统会将CPU切换到另一个线程。这种切换需要时间，而且切换回来的时候，之前的线程可能需要重新加载数据到CPU缓存中。
缓存失效： 当一个线程释放锁之后，另一个线程获取锁并修改共享数据，会导致其他线程的CPU缓存失效，需要重新从内存中加载数据。
优先级反转： 如果一个高优先级的线程等待一个低优先级的线程释放锁，而低优先级的线程又被其他线程抢占，那么高优先级的线程就被迫等待，导致优先级反转。

简单来说，锁竞争就像交通堵塞，大家都堵在路上，谁也走不了。

第三幕：无锁数据结构，釜底抽薪的妙招

既然锁竞争这么讨厌，那能不能不用锁呢？答案是肯定的！无锁数据结构就是一种不用锁也能保证线程安全的数据结构。

无锁数据结构通常使用原子操作来实现，原子操作是不可分割的操作，要么全部执行，要么全部不执行。就好比一个原子弹，要么爆炸，要么不爆炸，没有中间状态。

#include <iostream>
#include <thread>
#include <atomic>

std::atomic<int> atomic_data(0); // 原子整数

void atomic_increment() {
    for (int i = 0; i < 100000; ++i) {
        atomic_data++; // 原子加操作
    }
}

int main() {
    std::thread t1(atomic_increment);
    std::thread t2(atomic_increment);

    t1.join();
    t2.join();

    std::cout << "Atomic data: " << atomic_data << std::endl; // 预期输出: 200000
    return 0;
}

这段代码使用了std::atomic<int>来存储共享数据，并使用原子加操作atomic_data++来增加数据。这样就避免了锁的使用，提高了程序的性能。

当然，无锁数据结构也不是万能的。它需要仔细的设计和实现，否则容易出现各种bug。而且，有些复杂的数据结构和操作很难用无锁的方式来实现。

第四幕：原子操作，精打细算的工具

原子操作是实现无锁数据结构的基础。C++11提供了丰富的原子操作，包括：

原子加载 (load): 原子地读取一个值。
原子存储 (store): 原子地写入一个值。
原子交换 (exchange): 原子地将一个新值写入到原子变量，并返回旧值。
原子比较并交换 (compare_exchange_weak/strong): 原子地比较原子变量的值和一个预期值，如果相等，则将原子变量的值设置为新值。

#include <iostream>
#include <atomic>
#include <thread>

std::atomic<int> counter(0);

void increment_counter() {
  for (int i = 0; i < 1000; ++i) {
    int expected = counter.load();
    int desired = expected + 1;

    // 使用 weak 版本，性能更好，但可能需要循环重试
    while (!counter.compare_exchange_weak(expected, desired)) {
      // expected 可能会被其他线程修改，需要重新加载
      expected = counter.load();
      desired = expected + 1;
    }
  }
}

int main() {
  std::thread t1(increment_counter);
  std::thread t2(increment_counter);

  t1.join();
  t2.join();

  std::cout << "Counter value: " << counter << std::endl; // 预期输出: 2000
  return 0;
}

这段代码使用了compare_exchange_weak来实现原子计数器。compare_exchange_weak会比较counter的值和expected的值，如果相等，则将counter的值设置为desired。如果counter的值和expected的值不相等，则compare_exchange_weak会失败，需要重新加载counter的值并重试。

compare_exchange_weak和compare_exchange_strong的区别在于，compare_exchange_weak可能会在即使原子变量的值没有改变的情况下也返回失败，因此需要在一个循环中重试。而compare_exchange_strong只有在原子变量的值确实改变的情况下才会返回失败。一般来说，compare_exchange_weak的性能更好，因为它允许编译器生成更优化的代码。

第五幕：无锁数据结构的实战演练

说了这么多理论，让我们来看几个无锁数据结构的例子：

无锁队列 (Lock-Free Queue): 一种可以在多个线程之间安全地传递数据的队列，不用锁也能保证线程安全。

#include <iostream>
#include <thread>
#include <atomic>
#include <memory>

template <typename T>
class LockFreeQueue {
private:
    struct Node {
        std::shared_ptr<T> data;
        Node* next;
    };

    std::atomic<Node*> head;
    std::atomic<Node*> tail;

public:
    LockFreeQueue() : head(new Node{nullptr, nullptr}), tail(head.load()) {}

    void enqueue(T value) {
        Node* newNode = new Node{std::make_shared<T>(value), nullptr};
        Node* tailNode = tail.load();
        Node* nextNode = nullptr;

        // 尝试将新节点链接到队列尾部
        while (!tail.compare_exchange_weak(tailNode, newNode)) {
            tailNode = tail.load();
        }

        // 尝试将新节点链接到队列尾部 (如果其他线程已经修改了 tail)
        tailNode->next = newNode;
    }

    std::shared_ptr<T> dequeue() {
        Node* headNode = head.load();
        Node* nextNode = headNode->next;

        if (nextNode == nullptr) {
            return nullptr; // 队列为空
        }

        // 尝试更新 head 指针
        if (head.compare_exchange_strong(headNode, nextNode)) {
            std::shared_ptr<T> data = nextNode->data;
            delete headNode; // 释放旧的 head 节点
            return data;
        } else {
            return nullptr; // 其他线程已经修改了 head
        }
    }
};

int main() {
    LockFreeQueue<int> queue;

    std::thread t1([&]() {
        for (int i = 0; i < 10; ++i) {
            queue.enqueue(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    std::thread t2([&]() {
        for (int i = 0; i < 10; ++i) {
            std::shared_ptr<int> value = queue.dequeue();
            if (value) {
                std::cout << "Dequeued: " << *value << std::endl;
            } else {
                std::cout << "Queue is empty." << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(15));
        }
    });

    t1.join();
    t2.join();

    return 0;
}

这个无锁队列使用链表来实现，head和tail分别指向队列的头部和尾部。enqueue操作将新节点添加到队列尾部，dequeue操作从队列头部移除节点。

无锁栈 (Lock-Free Stack): 一种可以在多个线程之间安全地push和pop数据的栈，不用锁也能保证线程安全。

#include <iostream>
#include <thread>
#include <atomic>
#include <memory>

template <typename T>
class LockFreeStack {
private:
    struct Node {
        std::shared_ptr<T> data;
        Node* next;
    };

    std::atomic<Node*> head;

public:
    LockFreeStack() : head(nullptr) {}

    void push(T value) {
        Node* newNode = new Node{std::make_shared<T>(value), head.load()};

        // 尝试将新节点设置为栈顶
        while (!head.compare_exchange_weak(newNode->next, newNode)) {
            newNode->next = head.load(); // 如果失败，重新加载 head
        }
    }

    std::shared_ptr<T> pop() {
        Node* headNode = head.load();

        if (headNode == nullptr) {
            return nullptr; // 栈为空
        }

        Node* nextNode = headNode->next;

        // 尝试更新 head 指针
        if (head.compare_exchange_strong(headNode, nextNode)) {
            std::shared_ptr<T> data = headNode->data;
            delete headNode; // 释放旧的 head 节点
            return data;
        } else {
            return nullptr; // 其他线程已经修改了 head
        }
    }
};

int main() {
    LockFreeStack<int> stack;

    std::thread t1([&]() {
        for (int i = 0; i < 10; ++i) {
            stack.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    std::thread t2([&]() {
        for (int i = 0; i < 10; ++i) {
            std::shared_ptr<int> value = stack.pop();
            if (value) {
                std::cout << "Popped: " << *value << std::endl;
            } else {
                std::cout << "Stack is empty." << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(15));
        }
    });

    t1.join();
    t2.join();

    return 0;
}

这个无锁栈使用链表来实现，head指向栈顶。push操作将新节点添加到栈顶，pop操作从栈顶移除节点。

第六幕：原子操作的优先级，选择合适的工具

在使用原子操作时，我们需要根据具体的需求选择合适的操作。一般来说，优先级如下：

无锁数据结构优先： 如果可以使用无锁数据结构来解决问题，那么就尽量使用无锁数据结构。
原子变量优先： 如果需要对单个变量进行原子操作，那么就使用原子变量。
原子函数优先： 如果需要对多个变量进行原子操作，那么就使用原子函数。

表格总结:

技术方案	优点	缺点	适用场景
锁	简单易懂，容易实现	锁竞争导致性能下降，上下文切换开销，优先级反转	简单的数据共享，低并发场景
无锁数据结构	避免锁竞争，提高性能	设计复杂，容易出错，需要仔细的测试和验证	高并发，对性能要求高的场景
原子操作	轻量级，避免锁竞争	需要手动管理内存和同步，容易出现 ABA 问题	简单的原子操作，例如计数器，标志位
CAS (compare and swap)	实现无锁数据结构的关键，灵活	容易导致 ABA 问题，需要循环重试	实现无锁数据结构，例如无锁队列，无锁栈

第七幕：ABA 问题，隐形的陷阱

在使用compare_exchange时，需要注意ABA问题。ABA问题是指，一个变量的值从A变成了B，又从B变回了A。这时，compare_exchange会认为变量的值没有改变，从而导致错误。

举个例子，假设有一个指针指向一个对象，线程A读取了这个指针的值A，然后线程B将这个指针的值从A改成了B，又改回了A。这时，线程A使用compare_exchange来更新这个指针的值，会发现指针的值仍然是A，从而认为没有其他线程修改过这个指针，从而导致错误。

为了解决ABA问题，可以使用版本号或者双指针等技术。

第八幕：总结与展望

今天我们聊了C++中减少锁竞争的方法，主要包括无锁数据结构和原子操作。无锁数据结构可以避免锁竞争，提高程序的性能。原子操作是实现无锁数据结构的基础。

当然，无锁编程也不是银弹，它需要仔细的设计和实现，否则容易出现各种bug。而且，有些复杂的数据结构和操作很难用无锁的方式来实现。

未来，随着硬件的发展和编程技术的进步，无锁编程将会越来越重要。希望今天的讲座能帮助大家更好地理解和使用无锁编程，编写出更高效、更可靠的C++程序。

感谢大家的聆听！

C++ 减少锁竞争：无锁数据结构与原子操作的优先级 (讲座版)

发表回复 取消回复

发表回复取消回复