关于C/C++轻量级HTTP协议解析项目需要注意的几个关键实现

1. 基于小顶堆的计时器

其实这个之前有接触过，学习select的时候超时关闭就用到了这部分的知识，但是我们那个时候用的是vector，每一次都需要O(n)的遍历，时间效率极差，并不能很好的处理高并发的场景，为此，才有了小顶堆的计时器来作为优化。

关于这个我想说的是对于heap来说，我么们不能把他当作一个数组来看，我们的把他当作一个树来看，也就是所谓的堆，不然，又要变成O(n)的时间复杂度了，我们定义，某一个父节点i，其i * 2 + 1为左节点，i * 2 + 2为右节点，这里要注意，我们的下标是从0开始的，也就是说，寻找左右节点的结果都是下标，这是为了方便进行上浮和下沉两次操作，因为每当我交换或删除数据的时候，堆的结构就会改变，我们必须采用上浮和下沉操作来恢复堆的结构。而下沉时，我们需要对父节点的左右节点进行比较以此交换，上浮只需要对父节点进行比较，对于某一个子节点i，它的父节点为(i - 1) / 2。这个其实可以推出来这是左节点的父节点，右节点是(i - 2) / 2，但是右节点也要用这个公式，因为整数除法无论i取任意值得到的其父节点的下标永远一样。但是我们为什么不直接sort呢？还是时间复杂度的问题，sort的时间复杂度为o(nlogn)。

其他就没什么好说的了，就感觉手写堆还是比较难的吧。

1. 线程安全队列

该说不说，这个确实有点难，就单看这个shutdown_我就搞不明白，为什么pop函数的cv的条件是队列不为空或者shutdown_为true的时候才弹出数据，其实队列不空确实是一个基础，但是shutdown_为true是什么意思？都关闭了还能弹？其实我们用现实的例子来看就行了，比如你开了一家奶茶店，你的标准就是服务好每一个人，那么有一天，奶茶店所在的地区刮起了大风！这个时候，为了不让店内财产受损失，你下定决心关门（shutdown_ = true）

但是问题来了，还有三份奶茶还没做（queue_.size() = 3），这三位客人等了好久了就想尝尝你们家的奶茶，这不行啊，这三个任务必须完成！但是已经下定决心关门了，就意味着，完成剩下的所有任务，完成后，不再接受新任务。这也就是为什么当shutdown_为true时，必须检查队列里是否还有数据，如有，立刻弹出完成任务，如果没有，也就意味着全部处理完了，先行判断if(queue_.empty() && shutdown_) return false;这就是所有的流程，说实话还挺精妙的，尤其是不熟悉cv(condition_variable)的，确实很难懂很难懂，建议去深度学学cv，尤其是线程的领域里，cv是非常重要的一个点。

代码:

mini_heap_timer.h

#ifndef MINI_HEAP_TIMER_H #define MINI_HEAP_TIMER_H #include <vector> #include <unordered_map> #include <functional> #include <mutex> #include <ctime> class min_heap_timer{ public: min_heap_timer() = default; ~min_heap_timer() = default; void add(int fd,int timeout_sec, std::function<void()> callback = nullptr); void refresh(int fd,int timeout_sec); void remove(int fd); void tick(); private: struct Timernode { int fd; time_t expire; std::function<void()> callback; bool operator<(const Timernode& other) const { return expire > other.expire; } }; void sift_up(size_t i); void sift_down(size_t i); void swap_node(size_t i,size_t j); std::vector<Timernode> heap; std::unordered_map<int,size_t> fd_to_heap; std::mutex mtx; }; #endif

mini_heap_timer.cpp

#include "mini_heap_timer.h" #include <stdexcept> void min_heap_timer::add(int fd,int timeout_sec, std::function<void()> callback){ std::lock_guard<std::mutex> lock(mtx); auto it = fd_to_heap.find(fd); if(it != fd_to_heap.end()){ size_t i = it -> second; if(i < heap.size()){ swap_node(i,heap.size() - 1); heap.pop_back(); fd_to_heap.erase(it); if(i < heap.size()){ sift_up(i); sift_down(i); } } } time_t expire = time(nullptr) + timeout_sec; heap.push_back({fd,expire,std::move(callback)}); size_t idx = heap.size() - 1; fd_to_heap[fd] = idx; sift_up(idx); } void min_heap_timer::refresh(int fd,int timeout_sec){ std::lock_guard<std::mutex> lock(mtx); auto it = fd_to_heap.find(fd); if(it != fd_to_heap.end()){ int i = it -> second; int new_timeout = time(nullptr) + timeout_sec; heap[i].expire = new_timeout; sift_up(i); sift_down(i); }else{ add(fd,timeout_sec,nullptr); } } void min_heap_timer::remove(int fd){ std::lock_guard<std::mutex> lock(mtx); auto it = fd_to_heap.find(fd); if(it != fd_to_heap.end()){ size_t i = it -> second; if(i < heap.size()){ swap_node(i,heap.size() - 1); heap.pop_back(); fd_to_heap.erase(it); if(i < heap.size()){ sift_up(i); sift_down(i); } } }else return; } void min_heap_timer::tick(){ std::lock_guard<std::mutex> lock(mtx); time_t now = time(nullptr); while(!heap.empty() && heap[0].expire <= now){ Timernode node = heap[0]; fd_to_heap.erase(node.fd); if(heap.size() > 1){ swap_node(0,heap.size() - 1); heap.pop_back(); sift_down(0); }else{ heap.pop_back(); } if(node.callback != nullptr){ node.callback(); } } } void min_heap_timer::sift_up(size_t i){ while(i > 0){ size_t parent = (i - 1) / 2; if(heap[i] < heap[parent]){ swap_node(i,parent); i = parent; }else break; } } void min_heap_timer::sift_down(size_t i){ size_t n = heap.size(); while(true){ size_t left = 2 * i + 1; size_t right = 2 * i + 2; size_t smallest = i; if(left < n && heap[left] < heap[smallest]) smallest = left; if(right < n && heap[right] < heap[smallest]) smallest = right; if(smallest != i){ swap_node(i,smallest); i = smallest; }else break; } } void min_heap_timer::swap_node(size_t i, size_t j){ std::swap(heap[i],heap[j]); fd_to_heap[heap[i].fd] = i; fd_to_heap[heap[j].fd] = j; }

thread_safe_queue.h

#ifndef THREAD_SAFE_QUEUE_H #define THREAD_SAFE_QUEUE_H #include <queue> #include <mutex> #include <condition_variable> #include <chrono> template <typename T> class ThreadSafeQueue { public: ThreadSafeQueue() = default; ~ThreadSafeQueue(){ shutdown(); } ThreadSafeQueue(const ThreadSafeQueue&) = delete; ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete; void push(const T& item){ std::lock_guard<std::mutex> lock(mtx); if(shutdown_) return; queue_.push(item); cv.notify_one(); } void push(T&& item){ std::lock_guard<std::mutex> lock(mtx); if(shutdown_) return; queue_.push(item); cv.notify_one(); } bool pop(T& item){ std::unique_lock<std::mutex> lock(mtx); cv.wait(lock,[this](){return !queue_.empty() || shutdown_;}); if (queue_.empty() && shutdown_) { return false; } item = std::move(queue_.front()); queue_.pop(); return true; } bool pop(T& item,int timeout_ms){ std::unique_lock<std::mutex> lock(mtx); if(!cv.wait_for(lock,std::chrono::milliseconds(timeout_ms), [this](){return !queue_.empty() || shutdown_;})){ return false; } if(queue_.empty() && shutdown_) return false; item = std::move(queue_.front()); queue_.pop(); return true; } void shutdown(){ std::lock_guard<std::mutex> lock(mtx); shutdown_ = true; cv.notify_all(); } bool empty() const{ std::lock_guard<std::mutex> lock(mtx); return queue_.empty(); } size_t size() const { std::lock_guard<std::mutex> lock(mtx); return queue_.size(); } private: std::queue<T> queue_; mutable std::mutex mtx; std::condition_variable cv; bool shutdown_ = false; }; #endif