经典并发数据结构手写

一、手写 MPSC（多写一读）阻塞队列

这是异步日志库的核心结构——多个生产者（业务线程）写日志，一个消费者（刷盘线程）读日志。

难点：多写需要保护，单读不需要。

整体操作流程：

push（多生产者）：
1. 获取互斥锁，保护队列的并发访问。
2. 如果队列已停止，丢弃数据。
3. 将数据推入队列。
4. 释放锁后调用 notify_one 唤醒消费者（解锁后再通知，减少线程唤醒时的锁争用）。
pop（单消费者）：
1. 获取互斥锁。
2. 等待条件：队列不为空或队列已停止（cv.wait 自动释放/重获锁）。
3. 如果已停止且队列为空，返回默认值（或抛异常）。
4. 取出队首元素并返回。
stop：
1. 获取锁，设置停止标志。
2. notify_all 唤醒消费者线程，使其从 wait 中退出。

template <typename T>
class MPSCBlockingQueue {
    std::queue<T> queue_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stopped_ = false;

public:
    void push(T item) {
        {
            std::lock_guard<std::mutex> lock(mtx_);// ① 获取锁（多生产者互斥）
            if (stopped_) return;                  // ② 已停止则丢弃
            queue_.push(std::move(item));           // ③ 入队
        }                                          // ④ 解锁
        cv_.notify_one();                          // ⑤ 唤醒消费者（解锁后 notify 减少竞争）
    }

    T pop() {
        std::unique_lock<std::mutex> lock(mtx_);   // ① 获取锁
        cv_.wait(lock, [this] {                    // ② 等待条件：有数据或已停止
            return stopped_ || !queue_.empty();
        });

        if (stopped_ && queue_.empty()) {
            return T{};                            // ③ 已停止且无数据 → 返回空值
        }

        T item = std::move(queue_.front());         // ④ 取出队首元素
        queue_.pop();
        return item;                                // ⑤ 返回数据
    }

    void stop() {
        {
            std::lock_guard<std::mutex> lock(mtx_);// ① 获取锁
            stopped_ = true;                        // ② 设置停止标志
        }
        cv_.notify_all();                           // ③ 唤醒所有等待的线程
    }
};

为什么叫 MPSC？

M（Multi）：push 可以被多个线程同时调用（通过 mutex 保护）。
S（Single）：pop 只能被一个线程调用（不需要保护，因为就一个消费者）。
C（Consumer）：消费者线程消费数据。

性能特点：

push 在竞争激烈时互斥锁是瓶颈（多线程抢同一把锁）。
改进方向：用 无锁 SPSC 队列（你的 SimpleAsyncLogger 用的就是这思路）。
带宽大于无锁方案（因为单消费者不需要 CAS），但延迟抖动更严重（锁争用）。

二、手写线程安全的 LRU（Least Recently Used）缓存（LeetCode 146）

核心思想：缓存空间有限，放不下所有数据。当满了要淘汰时，淘汰最久没被用过的那一个——因为最近用过的很可能马上还要用。

功能：

get(key)：存在返回 value，不存在返回 -1。
put(key, value)：不存在则插入；如果容量满，淘汰最近最少使用的。

数据结构选择：

用 unordered_map 实现 $O(1)$ 查找。
用 双向链表 维护使用顺序（最近使用的在头部，最久未使用的在尾部）。
map 的 value 存 list 的迭代器，实现 $O(1)$ 移动到头部。

整体操作流程：

get(key)：
1. 在 unordered_map 中查找 key。
2. 不存在 → 返回 -1。
3. 存在 → 将该节点移到链表头部（表示最近使用过），返回对应的 value。
put(key, value)：
1. 在 unordered_map 中查找 key。
2. 已存在：更新 value，将节点移到链表头部。
3. 不存在：
  - 如果缓存已满（size >= capacity），淘汰链表尾部节点（最久未使用），从 map 中删除对应 key。
  - 在链表头部插入新节点 (key, value)，在 map 中记录 key → 迭代器的映射。

class LRUCache {
    int capacity_;
    list<pair<int, int>> items_;                     // 双向链表：头=最近使用，尾=最久未使用
    unordered_map<int, list<pair<int, int>>::iterator> cache_;// key → 链表节点的迭代器

public:
    LRUCache(int capacity) : capacity_(capacity) {}

    int get(int key) {
        auto it = cache_.find(key);                  // ① 在 map 中查找
        if (it == cache_.end()) return -1;            // ② 不存在 → 返回 -1

        items_.splice(items_.begin(), items_, it->second);// ③ 移到链表头部（最近使用）
        return it->second->second;                    // ④ 返回 value
    }

    void put(int key, int value) {
        auto it = cache_.find(key);

        if (it != cache_.end()) {                     // ① key 已存在
            it->second->second = value;               // ② 更新 value
            items_.splice(items_.begin(), items_, it->second);// ③ 移到链表头部
            return;
        }

        if (cache_.size() >= capacity_) {             // ④ 缓存已满
            auto& last = items_.back();               // ⑤ 取链表尾部（最久未使用）
            cache_.erase(last.first);                 // ⑥ 从 map 中删除
            items_.pop_back();                        // ⑦ 从链表中删除
        }

        items_.emplace_front(key, value);             // ⑧ 在链表头部插入
        cache_[key] = items_.begin();                 // ⑨ 在 map 中记录位置
    }
};

核心操作解析：

splice 是 std::list 的"搬家"操作——把节点从一个位置移到另一个位置，不分配内存、不拷贝数据、迭代器不失效：

// 把 it->second（一个 list 节点）从当前位置移到 items_.begin() 之前
items_.splice(items_.begin(), items_, it->second);
// 相当于 O(1) 时间把"刚访问的节点"移到链表头部

时间复杂度：get 和 put 都是 $O(1)$。 空间复杂度：$O(capacity)$。

多线程安全版本（在 get 和 put 外套 mutex）：

class ThreadSafeLRUCache {
    LRUCache cache_;
    std::mutex mtx_;

public:
    int get(int key) {
        std::lock_guard<std::mutex> lock(mtx_);
        return cache_.get(key);
    }

    void put(int key, int value) {
        std::lock_guard<std::mutex> lock(mtx_);
        cache_.put(key, value);
    }
};

三、面试追问 Q&A

Q1：为什么不用 shared_mutex 保护 LRU？ A：get 也会修改链表（把访问的节点移到头部），不是纯读操作。用 shared_mutex 的读锁不行，因为读操作也有写（修改链表顺序）。如果有"读但不修改顺序"的版本，可以用 shared_mutex 优化。

Q2：splice 为什么是 $O(1)$？ A：list 的 splice 只是调整 3 对 prev/next 指针的指向。不分配内存，不调用拷贝构造。对比 vector 在中间插入需要搬移 $O(N)$ 个元素。

Q3：为什么 LRU 不用单链表？ A：因为"移到头部"需要先找到前驱节点修改指针，单链表查找前驱需要 $O(N)$。

一、手写 MPSC（多写一读）阻塞队列#

二、手写线程安全的 LRU（Least Recently Used） 缓存（LeetCode 146）#

三、面试追问 Q&A#

一、手写 MPSC（多写一读）阻塞队列

二、手写线程安全的 LRU（Least Recently Used）缓存（LeetCode 146）

三、面试追问 Q&A