[项目深挖]仿muduo库的并发服务器的解析与优化方案

标题：[项目深挖]仿muduo库的并发服务器的优化方案 @水墨不写bug

文章目录

• 一、buffer 模块
• （1）线性缓冲区+直接扩容—->环形缓冲区+定时扩容（只会扩容一次）
• （2）使用双缓冲（Double Buffering）
• （3）数据丢弃策略
• • 为什么视频传输选择不可靠的UDP协议？
• （4）零拷贝
• • 为什么零拷贝重要？
  • 零拷贝的典型场景
  • 传统数据传输的过程
  • 零拷贝的过程
  • 实现零拷贝的技术
  • • 1. `sendfile` 系统调用
    • 2. `mmap` + `write`
    • 3. `splice` 系统调用
• 二、EventLoop 模块
• • 如何理解RunInLoop这一类的函数？
  • • 为什么需要 runInLoop？
    • runInLoop 的优点
  • （1）减少锁的使用—-无锁队列
  • （2）优化定时器处理—-时间轮代替最小堆
  • • 为什么 TimerWheel 需要 weak_ptr？
• 三. ThreadPool 模块
• 四、Acceptor 模块
• • • 2. 关键组件
    • 3. 实现步骤
    • • 3.1. 定义负载均衡器
      • 3.2. 改造 `Acceptor` 模块
      • 3.3. 改造 `EventLoop` 模块
• 五、定时器模块
• 六、日志模块
• 七、 错误处理与监控
• • 1. 设计原则
  • 2. 实现步骤
  • • 2.1. 定义异常基类
    • 2.2. 定义派生异常类
    • 2.3. 在核心模块中抛出异常
    • 2.4. 全局捕获异常
  • 3. 优化与扩展
  • • 细化异常信息
    • 结合日志记录
    • 异常恢复机制
    • 支持自定义异常类型

一、buffer 模块

（1）线性缓冲区+直接扩容—->环形缓冲区+定时扩容（只会扩容一次）

如果缓冲区满了超过一定时间（10s）仍然处于高使用率（>=90%）的状态，则扩容一次，增大环形缓冲区的大小，后续不再扩容。扩容一次之后，关闭定时器。 基于历史数据自适应扩容：历史上缓冲区如果负载较高，可以选择较大扩容幅度；如果负载较低，可以选择较小扩容幅度。通过prev使用率与cur使用率求变化率。 如果扩容后的缓冲区仍然一直处于高使用率状态，则计入日志文件。 合理性： 缓冲区负载偶发性：当缓冲区满的情况是偶发的，而不是长期的瓶颈。 内存资源敏感性：扩容是有限制的（只扩容一次），避免了动态扩容带来的过多内存消耗。

（2）使用双缓冲（Double Buffering）

使用两个缓冲区，当一个缓冲区满时切换到另一个缓冲区，避免阻塞。通过状态变量控制两个缓冲区的切换。

（3）数据丢弃策略

当缓冲区满时，直接丢弃新到达的数据或旧数据。 丢弃最旧数据：移除环形缓冲区中最早的数据（比如日志系统中）。 丢弃新数据：直接丢弃当前要写入的内容（比如视频帧流中）。 优点： 避免系统阻塞，保证系统运行流畅。 缺点： 数据丢失可能会影响系统的业务逻辑。 适用场景： 应用对数据完整性要求不高，如日志、视频流等场景。

为什么视频传输选择不可靠的UDP协议？

UDP——无连接，不可靠，低延迟，面向数据报。 实时性 使用 TCP 时，丢包会触发重传机制，可能导致延迟增加或卡顿，不适合实时性要求高的场景。 UDP 没有重传机制，即使丢包，视频播放也不会被阻塞，用户可能只会看到短暂的画质下降。 容忍丢包 视频流通常使用编码技术（如 H.264、H.265），具有一定的抗丢包能力。 即使部分数据丢失，解码器仍然可以通过冗余信息或插值技术恢复画面，保证用户体验。 高效性 UDP 的开销比 TCP 更小，因为它没有复杂的连接管理、流量控制和拥塞控制。 对于带宽有限的网络环境，减少协议开销意味着可以传输更多的视频数据。 乱序容忍 视频播放有一定的缓冲区，可以通过序列号等方式重新排序数据包，解决 UDP 的乱序问题。 即使部分数据包延迟到达，也可以选择丢弃，而不会影响整体流畅度。

视频传输选择 UDP 的原因主要是为了满足实时性、高效性和丢包容忍的需求。尽管 UDP 本身是不可靠的，但结合应用层协议（如 RTP）、纠错技术（如 FEC）和优化手段（如自适应比特率），可以弥补其不足，确保视频流的质量和流畅性。对于实时性要求低的场景（如视频文件下载），则可以选择更可靠的 TCP。

（4）零拷贝

零拷贝（Zero-Copy） 是一种优化技术，旨在在计算机系统中减少数据复制的次数，以提高数据传输或处理的效率，尤其是在文件或网络数据的高效传输中。零拷贝的核心理念是避免 CPU 将数据从一个位置复制到另一个位置，而是通过特定的硬件或内核支持，直接在数据的生产者和消费者之间传递数据。

为什么零拷贝重要？

零拷贝的典型场景

传统数据传输的过程

以文件发送到网络为例，传统数据传输的步骤如下：

总共涉及 4 次数据拷贝，其中 CPU 负责完成至少 2 次数据复制。

零拷贝的过程

通过零拷贝技术，可以将上述过程优化为：

总共涉及 0 次用户态拷贝，CPU 只负责控制流程。

实现零拷贝的技术

以下是几种常见的零拷贝实现技术：

1. sendfile 系统调用

• 描述：
  • sendfile 是 Linux 提供的一种系统调用，用于将文件直接从内核页缓存发送到网络套接字。
• 工作原理：
  • 文件数据从磁盘被读取到内核页缓存后，直接从内核页缓存发送到网卡，无需经过用户态。
• 适用场景：
  • 文件服务器、Web 服务器等需要高效传输文件的场景。
• 示例代码：int fd = open("file.txt", O_RDONLY);
  int sock = socket(...);
  sendfile(sock, fd, NULL, file_size);

2. mmap + write

• 描述：
  • 使用 mmap 将文件映射到用户空间内存地址，然后直接调用 write 将数据发送到套接字。
• 工作原理：
  • 避免了从磁盘读取到用户缓冲区的额外拷贝。
• 适用场景：
  • 需要灵活访问文件内容，同时减少拷贝次数的场景。
• 示例代码：int fd = open("file.txt", O_RDONLY);
  char* data = mmap(NULL, file_size, PROT_READ, MAP_PRIVATE, fd, 0);
  write(sock, data, file_size);

3. splice 系统调用

• 描述：
  • splice 允许在两个文件描述符之间直接移动数据，减少拷贝。
• 工作原理：
  • 数据通过内核缓冲区直接从一个管道移动到另一个目标，无需用户态干预。
• 适用场景：
  • 网络数据流处理、文件复制等场景。
• 示例代码：int pipefd[2];
  pipe(pipefd);
  splice(file_fd, NULL, pipefd[1], NULL, file_size, SPLICE_F_MOVE);
  splice(pipefd[0], NULL, sock_fd, NULL, file_size, SPLICE_F_MOVE);

  ---

二、EventLoop 模块

EventLoop 是 Muduo 的核心模块之一，负责管理事件循环和分发。

如何理解RunInLoop这一类的函数？

在 Muduo 网络库中，runInLoop 这一类函数是为了在特定的线程（即事件循环线程）中执行某些任务而设计的。它们的主要用途是解决跨线程调用的问题，确保任务在正确的线程上下文中执行。

runInLoop 的主要作用是将一个任务（通常是回调函数）添加到当前的 EventLoop 中执行。如果调用 runInLoop 的线程不是事件循环的线程，那么任务会被放入事件循环的任务队列中，等待事件循环线程来执行。

为什么需要 runInLoop？

在多线程环境中，Muduo 的 EventLoop 是线程不安全的，即：不能直接从多个线程访问或修改同一个 EventLoop。 事件循环线程需要对事件进行处理，而其他线程可能需要向事件循环线程发起某些操作（例如注册回调函数、修改定时器等）。

问题： 如果直接在非事件循环线程中调用事件循环的操作，可能会导致数据竞争或崩溃。 解决： 使用 runInLoop 将任务安全地转移到事件循环线程中，确保任务在正确的线程上下文中被执行。 runInLoop 的实现原理

以下是 runInLoop 的主要工作机制：

1、判断线程上下文： 如果调用 runInLoop 的线程是当前 EventLoop 所属的线程，则直接执行任务。 如果调用线程不是事件循环线程，则将任务添加到任务队列中，等待事件循环线程来执行。

2、任务队列： EventLoop 内部有一个任务队列（通常是 std::vector<std::function<void()>>），用于存储需要在事件循环线程中执行的任务。

3、唤醒机制： 如果任务是从其他线程添加的，EventLoop 需要被唤醒（通常通过向 wakeupFd 写入数据），以便尽快处理新增的任务。

runInLoop 的优点

线程安全： 确保所有任务都在事件循环线程中执行，避免数据竞争。 高效唤醒： 使用轻量级唤醒机制（如 eventfd 或 pipe）快速响应任务。 任务聚合： 通过任务队列，可以批量处理任务，减少上下文切换。

（1）减少锁的使用—-无锁队列

问题：EventLoop 的跨线程操作（如 runInLoop 和 queueInLoop）使用了锁保护。 优化： 使用无锁队列（如基于 lock-free 的 CAS 算法）替代当前的 std::mutex。 在单线程场景下，完全移除锁。

无锁队列（Lock-Free Queue）是一种数据结构，在多线程环境中使用时，不需要依赖传统的互斥锁来同步线程间的访问，而是通过硬件支持的原子操作（如 CAS，Compare-And-Swap）来完成线程安全的操作。无锁队列通常具有更高的性能，因为它避免了锁的开销和可能的线程阻塞。

实现思路

使用 CAS 操作： CAS(ptr, old, new)：如果 *ptr == old，则将 *ptr 更新为 new，否则不更新，并返回是否成功。 记录队列头和尾： 使用原子变量指向队列的头部和尾部。 生产者操作（Enqueue）： 找到当前尾节点并尝试将新节点插入到尾部。 消费者操作（Dequeue）： 找到当前头节点并尝试移除它。

（2）优化定时器处理—-时间轮代替最小堆

问题：EventLoop 的定时器使用了最小堆存储，复杂度为 O(log N)，当定时任务量非常多时可能会产生性能瓶颈。 优化： 使用分层时间轮（TimerWheel）替代最小堆，降低复杂度到 O(1)，尤其适合高频定时任务场景。

使用智能指针管理定时任务 在 TimerWheel 中使用 shared_ptr 和 weak_ptr 来管理定时任务是一种常见的设计，主要目的是解决 内存管理 和 资源生命周期控制 的问题。 shared_ptr 和 weak_ptr 的基本概念

shared_ptr： 一个智能指针，提供共享所有权。 当最后一个 shared_ptr 被销毁时，所管理的对象会自动释放。 使用 use_count() 方法可以查看当前有多少个 shared_ptr 在共享同一对象。

weak_ptr： 一个不影响引用计数的智能指针。 只能通过 lock() 方法访问所管理的对象。 当所指向的对象被销毁时，weak_ptr 会变为无效（即 expired() 返回 true）。

为什么 TimerWheel 需要 weak_ptr？

在 TimerWheel 中，一个定时任务可能需要被多个地方引用，例如：

TimerWheel 的槽位：每个槽位可能存储一组任务（通常是 shared_ptr < TimerTask > ）。 用户代码：用户可能直接持有某个定时任务的引用，以便随时取消或修改任务。

如果只使用 shared_ptr，会导致循环引用的问题。例如：

定时任务本身持有引用，而 TimerWheel 的槽位又持有定时任务的 shared_ptr。 这种情况下，shared_ptr 的引用计数永远不会降为 0，导致内存泄漏。

为了解决这种问题，使用 weak_ptr 来打破循环引用：

TimerWheel 的槽位使用 weak_ptr 存储定时任务。 用户持有的 shared_ptr 决定了任务的生命周期。

weak_ptr 的作用

1.避免循环引用： 如果定时任务被 shared_ptr 引用，但槽位只保留了 weak_ptr，当用户的 shared_ptr 被销毁时，任务会自动释放，避免了内存泄漏。

2.弱引用机制： TimerWheel 的槽位只需要一个弱引用来跟踪定时任务，而不需要管理其生命周期。 在执行定时任务时，可以通过 weak_ptr::lock() 检查任务是否仍然有效。如果任务已被用户取消或销毁，则无需执行。

3.任务销毁的灵活性： 用户可以随时销毁 shared_ptr，从而取消任务。 同时，TimerWheel 的槽位不会影响任务的生命周期。

三. ThreadPool 模块

ThreadPool 模块用于管理线程池，处理多线程任务。

四、Acceptor 模块

Acceptor 模块负责监听新连接并分发给 TcpConnection。

以下是实现动态负载均衡的设计方案和步骤：

2. 关键组件

3. 实现步骤

3.1. 定义负载均衡器

创建一个负载均衡器类，负责跟踪线程的负载信息并选择合适的线程。

#include <vector>
#include <memory>
#include <mutex>
#include <functional>

class EventLoop; // 前向声明

class LoadBalancer {
public:
LoadBalancer() = default;

// 添加一个线程的负载监控
void addEventLoop(EventLoop* loop) {
std::lock_guard<std::mutex> lock(mutex_);
eventLoops_.emplace_back(loop, 0); // 初始负载为 0
}

// 更新线程的负载（比如连接数变化）
void updateLoad(EventLoop* loop, int delta) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& [eventLoop, load] : eventLoops_) {
if (eventLoop == loop) {
load += delta;
break;
}
}
}

// 获取负载最轻的线程
EventLoop* getLeastLoadedEventLoop() {
std::lock_guard<std::mutex> lock(mutex_);
EventLoop* bestLoop = nullptr;
int minLoad = INT_MAX;

for (const auto& [eventLoop, load] : eventLoops_) {
if (load < minLoad) {
bestLoop = eventLoop;
minLoad = load;
}
}

return bestLoop;
}

private:
std::vector<std::pair<EventLoop*, int>> eventLoops_; // 每个线程及其负载
std::mutex mutex_; // 保护线程安全
};

3.2. 改造 Acceptor 模块

修改 Acceptor，在接收到新连接时调用负载均衡器，选择最优线程。

#include "LoadBalancer.h"
#include "EventLoop.h"
#include <functional>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <iostream>

class Acceptor {
public:
Acceptor(EventLoop* baseLoop, int port, LoadBalancer& loadBalancer)
: baseLoop_(baseLoop), loadBalancer_(loadBalancer) {
// 创建监听套接字
listenFd_ = ::socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);

// 绑定地址和端口
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(port);
::bind(listenFd_, (sockaddr*)&addr, sizeof(addr));

// 开始监听
::listen(listenFd_, SOMAXCONN);
}

~Acceptor() {
::close(listenFd_);
}

// 开始接受连接
void acceptConnections() {
while (true) {
sockaddr_in clientAddr{};
socklen_t clientLen = sizeof(clientAddr);
int connFd = ::accept4(listenFd_, (sockaddr*)&clientAddr, &clientLen, SOCK_NONBLOCK);

if (connFd < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
break; // 无更多连接
} else {
std::cerr << "Accept error: " << strerror(errno) << std::endl;
break;
}
}

// 使用负载均衡器选择最优线程
EventLoop* targetLoop = loadBalancer_.getLeastLoadedEventLoop();
if (targetLoop) {
// 将新连接分配给目标线程
targetLoop->runInLoop([connFd, targetLoop]() {
targetLoop->addConnection(connFd);
});

// 更新负载信息
loadBalancer_.updateLoad(targetLoop, 1);
} else {
std::cerr << "No available EventLoop to handle connection" << std::endl;
::close(connFd);
}
}
}

private:
EventLoop* baseLoop_; // 主线程的事件循环
int listenFd_; // 监听套接字
LoadBalancer& loadBalancer_; // 负载均衡器
};

3.3. 改造 EventLoop 模块

扩展 EventLoop，支持添加和处理新连接。

#include <functional>
#include <vector>
#include <mutex>
#include <unistd.h>
#include <iostream>

class EventLoop {
public:
EventLoop() = default;

void runInLoop(const std::function<void()>& task) {
// 简化的事件循环任务执行
std::lock_guard<std::mutex> lock(mutex_);
tasks_.emplace_back(task);
}

void addConnection(int connFd) {
std::cout << "Handling new connection on EventLoop: " << this << std::endl;
connections_.emplace_back(connFd);
}

void processTasks() {
std::vector<std::function<void()>> tasksCopy;
{
std::lock_guard<std::mutex> lock(mutex_);
tasksCopy.swap(tasks_);
}

for (const auto& task : tasksCopy) {
task();
}
}

private:
std::vector<std::function<void()>> tasks_; // 待处理任务
std::mutex mutex_; // 保护任务队列
std::vector<int> connections_; // 当前连接列表
};

五、定时器模块

TimerQueue 模块用于管理定时任务。

分层时间轮的设计灵感来源于时钟的分层结构。例如：

时钟有秒针、分针、时针，每一层负责处理不同的时间粒度。 当秒针完成一圈（60 秒），会推动分针向前迈进一格。

类似的，分层时间轮将时间划分为多个层级，每一层是一个环形队列，队列中的每个槽（slot）存储对应时间间隔的定时任务。当时间轮的某一层转动一圈后，触发下一层的转动。 亮点：使用智能指针管理任务TimerTask 分层时间轮的结构

1、时间轮层级： 每一层是一个环形数组（类似时钟的轮盘）。 每个槽代表一个固定的时间间隔（时间粒度）。 第一层的时间粒度最小，越高层的时间粒度越大。

2、每个槽的内容： 每个槽存储定时任务的列表。 每个任务需要携带额外的元数据（例如任务到期时间）。

3、层级的关系： 第一层负责管理最小时间粒度的任务。 如果某个任务的时间超过当前层的最大时间范围，则被推进到下一级时间轮。

分层时间轮的工作原理 插入任务

根据任务的到期时间，计算任务所在的时间轮层级和槽位： 槽位索引 = ( 到期时间 , / , 时间粒度 ) , 如果任务的到期时间超过当前时间轮层级的时间范围，则将任务递归推入到更高层的时间轮。

定时器轮转

时间轮以固定的时间间隔轮转一格（类似秒针转动）。 每次轮转到一个槽位时，触发该槽中的任务。 如果某个任务的到期时间未到，则将其重新分配到更高层时间轮的对应槽位。

任务触发 当轮盘转动到任务所在的槽位，并且任务的到期时间与当前时间匹配时，触发任务。 分层时间轮的优势 高性能： 插入、删除、触发的时间复杂度接近 (O(1))。 适用于大规模定时任务的场景（例如实时系统的事件调度）。

灵活性： 支持多层时间轮，适应不同的时间范围和粒度需求。

内存效率： 由于采用环形数组存储任务，内存占用较小。

分层时间轮的应用场景

网络服务器： TCP 连接的超时管理。 应用于高性能网络库（如 Netty、Muduo 等）。 实时系统： 事件驱动的调度系统。 需要高效管理大量定时任务的场景。 分布式系统： 分布式任务调度（如分布式锁的过期时间管理）。 游戏引擎： 游戏中的倒计时、技能冷却等事件。

分层时间轮的局限性

时间粒度限制： 时间轮的粒度决定了定时器的精度，任务触发可能会有一定的延迟。 任务分层复杂性： 跨层任务需要递归推进，可能增加一定的实现复杂性。 非实时性： 高层时间轮的任务可能需要等待低层时间轮转动完成，导致延迟。

六、日志模块

Logging 模块是 Muduo 的日志系统。

七、 错误处理与监控

Muduo 缺乏对错误和性能的全面监控。

层级式异常管理 在 Muduo 库的基础上设计一个异常机制，可以通过引入基于 C++ 多态 的异常层次结构来分类和处理不同类型的异常。以下是具体的设计思路：

1. 设计原则

2. 实现步骤

2.1. 定义异常基类

创建一个通用的异常基类 MuduoException，它继承自 std::exception，并提供基本的异常信息接口。

#include <exception>
#include <string>

class MuduoException : public std::exception {
public:
explicit MuduoException(const std::string& message)
: message_(message) {}

// 返回异常信息
virtual const char* what() const noexcept override {
return message_.c_str();
}

// 提供异常类型的标识
virtual const char* type() const noexcept {
return "MuduoException";
}

protected:
std::string message_;
};

2.2. 定义派生异常类

为 Muduo 的核心模块定义具体的异常类型，这些类从 MuduoException 派生。

#include "MuduoException.h"

// 事件循环相关异常
class EventLoopException : public MuduoException {
public:
explicit EventLoopException(const std::string& message)
: MuduoException(message) {}

virtual const char* type() const noexcept override {
return "EventLoopException";
}
};

// TCP 连接相关异常
class TcpConnectionException : public MuduoException {
public:
explicit TcpConnectionException(const std::string& message)
: MuduoException(message) {}

virtual const char* type() const noexcept override {
return "TcpConnectionException";
}
};

// 定时器相关异常
class TimerQueueException : public MuduoException {
public:
explicit TimerQueueException(const std::string& message)
: MuduoException(message) {}

virtual const char* type() const noexcept override {
return "TimerQueueException";
}
};

2.3. 在核心模块中抛出异常

在 Muduo 的核心模块中，当检测到错误情况时，抛出对应的异常。

示例：在 EventLoop 模块中抛出异常

#include "DerivedExceptions.h"
#include <iostream>

class EventLoop {
public:
void loop() {
try {
// 模拟事件循环错误
throw EventLoopException("Event loop encountered an error!");
} catch (const MuduoException& e) {
handleException(e);
}
}

private:
void handleException(const MuduoException& e) {
// 根据异常类型进行处理
if (std::string(e.type()) == "EventLoopException") {
std::cerr << "[EventLoop Error] " << e.what() << std::endl;
// 执行特定的恢复逻辑
} else {
std::cerr << "[Unknown Error] " << e.what() << std::endl;
}
}
};

2.4. 全局捕获异常

可以在应用的入口函数中统一捕获所有的 MuduoException 类型异常。

#include "EventLoop.cpp"

int main() {
try {
EventLoop loop;
loop.loop();
} catch (const MuduoException& e) {
std::cerr << "Caught a MuduoException: " << e.type() << " – " << e.what() << std::endl;
} catch (const std::exception& e) {
std::cerr << "Caught a std::exception: " << e.what() << std::endl;
} catch (...) {
std::cerr << "Caught an unknown exception" << std::endl;
}

return 0;
}

3. 优化与扩展

细化异常信息

• 为每个异常类型添加更多上下文信息（如错误码、模块名称、操作步骤等）。
• 示例：explicit TcpConnectionException(const std::string& message, int errorCode)
  : MuduoException(message), errorCode_(errorCode) {}

  int errorCode() const { return errorCode_; }

结合日志记录

• 在捕获异常后，将异常信息写入日志，方便后续排查问题。
• 示例：void logException(const MuduoException& e) {
  // 写入日志
  std::ofstream logFile("error.log", std::ios::app);
  logFile << "[" << e.type() << "] " << e.what() << std::endl;
  }

异常恢复机制

• 根据异常类型，尝试执行不同的恢复策略：
  • 重启事件循环。
  • 关闭并重建 TCP 连接。
  • 重新注册定时器。

支持自定义异常类型

• 提供一个工厂函数或宏，方便用户定义新的异常类型。

#define DEFINE_EXCEPTION(name, base) \\
class name : public base { \\
public: \\
explicit name(const std::string& message) \\
: base(message) {} \\
virtual const char* type() const noexcept { \\
return #name; \\
} \\
};

使用示例：

DEFINE_EXCEPTION(CustomException, MuduoException)

通过这种设计，可以在 Muduo 的基础上实现一个灵活、可扩展的异常机制，既满足了错误检测的需求，又增强了代码的可维护性和健壮性。

性能监控

• 增加性能监控接口，统计每个模块的延迟、吞吐量和错误率。

等待进一步更新与更正~