11、锁的优化技术：细粒度锁、锁分段（lock striping）、读写锁优化、无锁数据结构基础

各位同学，咱们今天聊点硬核的——锁的优化。说实话，我早年写多线程代码时，最怕的就是锁。不加锁吧，数据乱套；加锁吧，性能崩盘。后来踩的坑多了，慢慢摸索出一些门道。今天就把这些经验掰开揉碎了讲给你们听。

11.1 细粒度锁：把大锁拆成小锁

什么叫细粒度锁？说白了，就是把一把大锁拆成多把小锁。你想想看，一个商场只有一个大门，所有人进出都挤这一个门，那不得堵死？细粒度锁的思路就是多开几个门，让不同的人走不同的门。

举个实际例子。假设我们要实现一个线程安全的哈希表。最粗暴的做法是整个表加一把大锁：

// 粗粒度锁版本
class ThreadSafeHashMap {
    std::unordered_map<int, std::string> map_;
    std::mutex mutex_;
public:
    void insert(int key, const std::string& value) {
        std::lock_guard<std::mutex> lock(mutex_);
        map_[key] = value;
    }
    
    std::string find(int key) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = map_.find(key);
        return it != map_.end() ? it->second : "";
    }
};

这个版本有什么问题？所有操作串行化。哪怕两个线程操作的是完全不同的桶，也得排队等着。我在项目中遇到过这种情况：16核的机器，实际吞吐量还不如单线程。为什么？锁竞争太激烈了。

细粒度锁的做法是：每个桶或每组桶分配一把锁。

// 细粒度锁版本
class FineGrainedHashMap {
    static constexpr int BUCKET_COUNT = 256;
    std::unordered_map<int, std::string> buckets_[BUCKET_COUNT];
    std::mutex mutexes_[BUCKET_COUNT];
    
    int getBucketIndex(int key) {
        return std::hash<int>{}(key) % BUCKET_COUNT;
    }
    
public:
    void insert(int key, const std::string& value) {
        int idx = getBucketIndex(key);
        std::lock_guard<std::mutex> lock(mutexes_[idx]);
        buckets_[idx][key] = value;
    }
    
    std::string find(int key) {
        int idx = getBucketIndex(key);
        std::lock_guard<std::mutex> lock(mutexes_[idx]);
        auto it = buckets_[idx].find(key);
        return it != buckets_[idx].end() ? it->second : "";
    }
};

11.2 锁分段（Lock Striping）：Java 的 ConcurrentHashMap 是怎么做的

锁分段，其实就是细粒度锁的一种经典实现。我记得最早接触这个概念是在看 Java 的 ConcurrentHashMap 源码时。它的思路是：把整个数据结构分成多个段（Segment），每个段独立加锁。

为什么叫「分段」？你可以想象一个图书馆，分成多个阅览室。每个阅览室有自己的管理员（锁）。你要借书，只需要找对应阅览室的管理员，不用惊动整个图书馆。

在 C++ 中实现锁分段，我一般这样写：

template<typename Key, typename Value>
class StripedHashMap {
    static constexpr int STRIPE_COUNT = 16;
    
    struct StripedBucket {
        std::unordered_map<Key, Value> data;
        std::shared_mutex mutex;  // 读写锁
    };
    
    std::vector<StripedBucket> stripes_;
    
    int getStripeIndex(const Key& key) {
        return std::hash<Key>{}(key) % STRIPE_COUNT;
    }
    
public:
    StripedHashMap() : stripes_(STRIPE_COUNT) {}
    
    void insert(const Key& key, const Value& value) {
        int idx = getStripeIndex(key);
        std::unique_lock<std::shared_mutex> lock(stripes_[idx].mutex);
        stripes_[idx].data[key] = value;
    }
    
    Value find(const Key& key) {
        int idx = getStripeIndex(key);
        std::shared_lock<std::shared_mutex> lock(stripes_[idx].mutex);
        auto it = stripes_[idx].data.find(key);
        return it != stripes_[idx].data.end() ? it->second : Value{};
    }
};

11.3 读写锁优化：读多写少的场景

读写锁，说白了就是区分读操作和写操作。读操作可以并发，写操作必须独占。这个优化在「读多写少」的场景下效果特别明显。

我举个例子。假设你有一个配置管理系统，配置信息几乎不变，但成千上万个线程要读取。如果用普通互斥锁，所有读线程都得排队——这太浪费了。

C++17 提供了 std::shared_mutex，用法很简单：

class ConfigManager {
    std::map<std::string, std::string> config_;
    mutable std::shared_mutex rw_mutex_;
    
public:
    std::string get(const std::string& key) {
        std::shared_lock<std::shared_mutex> lock(rw_mutex_);
        auto it = config_.find(key);
        return it != config_.end() ? it->second : "";
    }
    
    void set(const std::string& key, const std::string& value) {
        std::unique_lock<std::shared_mutex> lock(rw_mutex_);
        config_[key] = value;
    }
};

看到区别了吗？get 用的是 shared_lock，多个线程可以同时持有；set 用的是 unique_lock，必须独占。

我曾经在一个日志系统中踩过这个坑。日志写入频繁，结果用了读写锁后，读线程几乎拿不到锁，系统响应时间飙升。后来改成无锁队列才解决问题。

11.4 无锁数据结构基础：CAS 和 ABA 问题

无锁编程，听起来很玄乎，其实核心就一句话：用原子操作代替锁。最常用的原子操作是 CAS（Compare-And-Swap）。

C++11 提供了 std::atomic，支持 CAS 操作：

std::atomic<int> counter{0};

// CAS 操作：如果当前值等于 expected，就更新为 desired
int expected = counter.load();
while (!counter.compare_exchange_weak(expected, expected + 1)) {
    // 如果 CAS 失败，expected 会被更新为当前值
    // 重试
}

无锁数据结构的基本思路是：用 CAS 循环代替锁。比如实现一个无锁栈：

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    
    std::atomic<Node*> head_{nullptr};
    
public:
    void push(const T& value) {
        Node* new_node = new Node{value, nullptr};
        new_node->next = head_.load();
        
        while (!head_.compare_exchange_weak(new_node->next, new_node)) {
            // 如果 head 被其他线程修改了，重试
        }
    }
    
    bool pop(T& value) {
        Node* old_head = head_.load();
        while (old_head && 
               !head_.compare_exchange_weak(old_head, old_head->next)) {
            // 重试
        }
        
        if (old_head) {
            value = old_head->data;
            delete old_head;
            return true;
        }
        return false;
    }
};

11.5 ABA 问题：无锁编程的经典陷阱

说到无锁编程，就不得不提 ABA 问题。这个问题我当年第一次遇到时，debug 了整整两天。

什么是 ABA 问题？假设线程 A 读取到指针 P 指向地址 X，然后线程 B 把 P 指向 Y，又改回 X。线程 A 的 CAS 检查发现 P 还是指向 X，就认为没有变化——但实际上中间已经变过了。

解决 ABA 问题的常见方法是使用带版本号的指针：

// 使用双字 CAS（DCAS）或 tagged pointer
struct TaggedPointer {
    Node* ptr;
    uint64_t tag;
};

std::atomic<TaggedPointer> head_;

// 每次修改都递增 tag
TaggedPointer old = head_.load();
TaggedPointer new_val{new_node, old.tag + 1};
while (!head_.compare_exchange_weak(old, new_val)) {
    new_val.ptr->next = old.ptr;
    new_val.tag = old.tag + 1;
}

11.6 如何选择：一张决策表

说了这么多，到底什么时候用哪种技术？我整理了一张表：

场景	推荐技术	原因
读多写少（读 > 90%）	读写锁	读操作可并发，性能提升明显
写操作频繁	细粒度锁或锁分段	避免写锁阻塞所有读操作
数据结构简单（栈、队列）	无锁数据结构	性能最好，但实现复杂
数据结构复杂（树、图）	细粒度锁	无锁实现过于复杂，容易出错
对延迟要求极高	无锁数据结构	避免锁导致的上下文切换

嗯，这里要注意：没有银弹。我见过太多人一上来就搞无锁编程，结果 bug 修了三个月。我的建议是：先从简单的锁开始，用性能分析工具（比如 perf）找到瓶颈，再针对性地优化。

11.7 知识体系总览

最后，我用一张图把今天的内容串起来：

这张图把今天讲的内容都串起来了。你可以看到，四种优化技术各有侧重，但核心目标是一致的：减少锁竞争，提高并发性能。

好了，关于锁的优化就讲到这里。记住我的一句话：先跑起来，再优化。不要一开始就追求极致性能，先把功能做对，然后用性能分析工具找瓶颈，最后针对性地应用这些技术。这样才是工程化的做法。

---

公众号:蓝海资料掘金营，微信deep3321