【C++】：使用 mutex + deque 实现一个简单的 LockedQueue

一、为什么需要线程安全队列在 C 中我们经常使用std::queue、std::deque等容器保存数据。例如std::dequeint queue; queue.push_back(1); queue.push_back(2);在单线程程序中这样使用没有什么问题但是如果多个线程同时操作同一个队列就可能产生数据竞争。例如现在有两个线程std::thread t1([]() { queue.push_back(1); }); std::thread t2([]() { queue.push_back(2); });两个线程可能同时进入push_back()共同修改queue内部的数据结构。需要注意的是std::deque std::queue std::vector这些 STL 容器本身并不会自动帮我们解决多个线程同时读写的问题。因此最直接的解决方案就是普通队列 ↓ 加入 mutex ↓ 每次操作队列之前先加锁 ↓ 操作结束后释放锁 ↓ 线程安全队列例如std::mutex mutex; std::dequeint queue;线程操作队列的时候mutex.lock(); queue.push_back(1); mutex.unlock();这样同一时刻只能有一个线程操作队列。假设线程1已经获得锁线程1获得 mutex ↓ push_back() ↓ 释放锁线程2此时也想操作队列线程2尝试获得 mutex ↓ 获取失败 ↓ 等待 ↓ 线程1释放 mutex ↓ 线程2获得 mutex这就是线程安全队列最基础的实现思路。二、LockedQueue 的基本结构可以把线程安全队列封装成一个模板类#include deque #include mutex templateclass T, typename StorageType std::dequeT class LockedQueue { private: std::mutex _lock; // 保护队列的互斥锁 StorageType _queue; // 真正保存数据的容器 bool _canceled; // 队列是否已经取消 public: LockedQueue() : _canceled(false) {} };这里最核心的其实只有两个成员std::mutex _lock; StorageType _queue;其中_queue负责保存数据而_lock负责保护_queue。模板templateclass T, typename StorageType std::dequeT表示这个队列保存的数据类型由T决定同时默认使用std::dequeT作为底层容器。例如LockedQueueint queue;此时内部实际上可以理解成std::dequeint _queue;如果创建LockedQueuestd::string queue;那么内部就是std::dequestd::string _queue;因此LockedQueue可以保存不同类型的数据。整个结构可以简单理解成LockedQueueint ┌─────────────────────┐ │ mutex _lock │ │ ↓ │ │ ┌─────────────────┐ │ │ │ 1 │ 2 │ 3 │ 4 │ │ │ └─────────────────┘ │ │ _queue │ └─────────────────────┘只要所有对_queue的操作都经过_lock就能够避免多个线程同时修改队列。三、实现线程安全的入队操作首先实现最基本的入队操作void add(const T item) { _lock.lock(); // 操作队列之前先加锁 _queue.push_back(item); // 将数据加入队尾 _lock.unlock(); // 操作完成后释放锁 }例如LockedQueueint queue; queue.add(10); queue.add(20); queue.add(30);队列中的数据就是队头 队尾 ↓ ↓ 10 → 20 → 30其中_queue.push_back(item);负责真正添加数据。而_lock.lock(); _lock.unlock();负责保证同一时刻只能有一个线程执行这段操作。假设两个线程同时调用queue.add(1); queue.add(2);实际执行过程可能是线程1 ↓ 获得锁 ↓ push_back(1) ↓ 释放锁 线程2 ↓ 获得锁 ↓ push_back(2) ↓ 释放锁因此两个线程不会同时修改_queue。不过在实际 C 开发中一般更加推荐使用 RAII 的方式管理锁void add(const T item) { std::lock_guardstd::mutex lock(_lock); _queue.push_back(item); }这里std::lock_guardstd::mutex lock(_lock);创建时自动执行加锁函数结束时自动释放锁。可以理解为进入函数 ↓ 创建 lock_guard ↓ mutex.lock() ↓ 操作 queue ↓ 函数结束 ↓ lock_guard 析构 ↓ mutex.unlock()与手动_lock.lock(); _queue.push_back(item); _lock.unlock();相比lock_guard最大的好处就是不用担心中途return或发生异常导致忘记释放锁。所以简单的临界区代码通常推荐std::lock_guardstd::mutex lock(_lock);四、实现线程安全的出队操作只有入队还不够还需要能够安全地从队列中取出数据。可以实现一个next()bool next(T result) { std::lock_guardstd::mutex lock(_lock); if (_queue.empty()) return false; result _queue.front(); // 取出队头元素 _queue.pop_front(); // 删除队头元素 return true; }这里的逻辑非常重要。首先std::lock_guardstd::mutex lock(_lock);获得互斥锁。然后判断if (_queue.empty()) return false;如果队列已经没有数据就直接返回false。如果有数据result _queue.front();先把队头元素保存到result中。然后_queue.pop_front();删除队头元素。最后return true;表示成功取到了数据。例如队列10 → 20 → 30执行int value; queue.next(value);首先value _queue.front();此时value 10然后_queue.pop_front();队列变成20 → 30因此next()实际上完成了获取队头元素 删除队头元素为什么必须把empty() front() pop_front()全部放在同一把锁里面因为如果分开加锁就可能出现这种情况线程1检查队列不为空 ↓ 释放锁 线程2取走最后一个元素 线程1继续 front()此时线程1再访问队头就会出现问题。所以if (_queue.empty()) return false; result _queue.front(); _queue.pop_front();应该看成一个完整操作整个过程都由同一把锁保护。这也是多线程编程中非常重要的一点不仅仅是单独保护某一行代码更重要的是保护一个完整的逻辑操作。有了next()之后就可以这样不断读取队列int value; while (queue.next(value)) { std::cout value std::endl; }只要队列里面还有数据queue.next(value)就返回true。当队列为空时返回false循环结束。五、多线程测试 LockedQueue现在把前面的内容组合起来实现一个简单的线程安全队列#include deque #include mutex templateclass T, typename StorageType std::dequeT class LockedQueue { private: std::mutex _lock; // 保护队列 StorageType _queue; // 保存数据 bool _canceled; // 取消标志 public: LockedQueue() : _canceled(false) {} // 添加元素 void add(const T item) { std::lock_guardstd::mutex lock(_lock); _queue.push_back(item); } // 获取并删除队头元素 bool next(T result) { std::lock_guardstd::mutex lock(_lock); if (_queue.empty()) return false; result _queue.front(); _queue.pop_front(); return true; } // 判断队列是否为空 bool empty() { std::lock_guardstd::mutex lock(_lock); return _queue.empty(); } // 取消队列 void cancel() { std::lock_guardstd::mutex lock(_lock); _canceled true; } // 判断队列是否已经取消 bool cancelled() { std::lock_guardstd::mutex lock(_lock); return _canceled; } };接下来创建两个生产线程向队列添加数据再创建两个线程读取数据#include iostream #include thread #include LockedQueue.h int main() { LockedQueueint queue; // 生产者线程1 std::thread producer1([]() { queue.add(1); queue.add(2); queue.add(3); queue.add(4); }); // 生产者线程2 std::thread producer2([]() { queue.add(5); queue.add(6); queue.add(7); queue.add(8); }); // 消费者线程1 std::thread consumer1([]() { int value; while (queue.next(value)) { std::cout std::this_thread::get_id() : pop value std::endl; } }); // 消费者线程2 std::thread consumer2([]() { int value; while (queue.next(value)) { std::cout std::this_thread::get_id() : pop value std::endl; } }); producer1.join(); producer2.join(); consumer1.join(); consumer2.join(); return 0; }这里一共创建了四个线程producer1 ──→ 添加 1 2 3 4 ──┐ │ ↓ LockedQueue ↑ │ producer2 ──→ 添加 5 6 7 8 ──┘ ↓ ┌────────┴────────┐ ↓ ↓ consumer1 consumer2 取出数据 取出数据两个生产者线程producer1 producer2不断调用queue.add();向队列中添加数据。两个消费者线程consumer1 consumer2不断调用queue.next();获取数据。因为add()和next()内部都使用了同一把std::mutex _lock;所以无论有多少线程操作这个队列同一时刻都只能有一个线程真正访问_queue。需要注意一个细节这个简单版本还有一个明显的问题while (queue.next(value))当消费者执行next()时如果队列刚好为空它会直接返回false。也就是说消费者发现队列为空 ↓ next() 返回 false ↓ 消费者线程结束但此时生产者有可能只是还没有来得及添加数据。例如消费者线程启动 ↓ 发现队列为空 ↓ 退出 ↓ 生产者线程 ↓ 添加数据这样生产者后面加入的数据就没人处理了。因此当前这个LockedQueue解决的是多个线程同时操作队列时的数据安全问题。但是它还没有解决队列为空时消费者应该如何等待生产者的问题。如果让消费者不断这样检查while (true) { if (queue.next(value)) { // 处理数据 } }又会产生大量无意义的循环占用 CPU。更好的办法是队列有数据 ↓ 消费者正常取数据 队列没数据 ↓ 消费者进入等待状态 生产者添加数据 ↓ 通知消费者 ↓ 消费者被唤醒 ↓ 继续取数据而实现这个过程就需要 C 多线程中的另一个重要工具std::condition_variable这也会进一步形成经典的生产者 ——→ 线程安全队列 ——→ 消费者生产者消费者模型。总结这一部分实现的LockedQueue本质上并不复杂它的核心结构就是std::deque std::mutex 简单线程安全队列其中最重要的几个操作是_queue.push_back(item); // 入队 _queue.front(); // 获取队头 _queue.pop_front(); // 删除队头而所有这些操作外面都通过std::mutex进行保护。其中std::lock_guardstd::mutex lock(_lock);利用 RAII 自动完成加锁和解锁是简单临界区中非常常见的写法。到这里解决的是多个线程同时操作队列 ↓ mutex 保证同一时间只有一个线程访问 ↓ 避免数据竞争但是还存在队列为空 ↓ 消费者应该怎么办这个问题。因此在线程安全队列的基础上下一步就可以继续加入std::condition_variable让消费者在队列为空时进入等待在生产者加入数据之后再将消费者唤醒从而实现一个更加完整的生产者消费者队列。0voice · GitHub