【项目日记】仿mudou的高并发服务器 --- 实现缓冲区模块，通用类型Any模块，套接字模块

2025-05-04 12:00:53 0 阅读

一个人知道自己为什么而活，

就可以忍受任何一种生活。

--- 尼采 ---

✨✨✨项目地址在这里 ✨✨✨

✨✨✨https://gitee.com/penggli_2_0/TcpServer✨✨✨

仿mudou的高并发服务器

1 主从Reactor模型
2 基础功能封装
- 2.1 缓冲区 Buffer模块
- 2.2 通用类型 Any类
- 2.3 套接字 Socket模块

1 主从Reactor模型

这个项目的目标是实现一个可以高效处理请求的服务器，那么对于这样的一个服务器要如何实现呢？

这里采取主从Reactor模型的方案：

主Reactor模型负责对监听套接字进行管理，进行获取连接操作。
从属Reactor负责对连接的数据进行处理，并且为了保证线程安全，每一个从属Reactor都要绑定一个线程，这个连接的任务都在这个线程中完成。

简单来说就是这样的一个模型：

其中管理连接的对象是Connection对象，这个类是高并发服务器的核心部分，这里包含了对套接字的处理，事件等待等一系列操作！而Connection对象中对于这些操作的管理也要通过其他的对象来进行！

主要的工作就是搭建起主从Reactor模型，但是这个模型不是一下子就可以写出来的。为了实现主从Reactor模型，我们先需要实现一些基础类，对基础功能进封装，然后对这些功能进行整合，最终实现主从Reactor模型！

2 基础功能封装

2.1 缓冲区 Buffer模块

Buffer模块是⼀个缓冲区模块，用于实现通信中用户态的接收缓冲区和发送缓冲区功能：

需要支持字符串读取/写入
需要支持char*缓冲区读取/写入
需要正常按行读取 — 方便解析http请求

Buffer模块成员变量很简单：

vector容器_buffer：对内存空间进行管理
uint64_t _reader_idx 读偏移：进行读取位置在_buffer容器中的偏移量，即读取的起始位置。
uint64_t _writer_idx 写偏移：下一次写入位置在_buffer容器中的偏移量，即写入的起始位置。

接下来实现一下基础功能：

构造函数：初始化读/写偏移为0 ，容器_buffer初始化一个大小 BUFFER_DEFAULT_SIZE。
获取当前写入起始地址：_buffer空间的起始地址加入写偏移量即写入起始地址。
获取当前读取起始地址： _buffer空间的起始地址加入读偏移量即读取起始地址。
获取缓冲区末尾空闲空间大小：写偏移之后的空闲空间，总体空间大小减去写偏移就是写偏移之后的空间大小。
获取缓冲区起始空闲空间大小：读偏移之前的空闲空间，其实就是读偏移的大小。
获取可读数据大小：写偏移减去读偏移就就之间可读空间的大小！
读/写偏移向后移动：
* 先根据len判断是否小于可读数据大小 len必须小于可读数据大小，然后移动读偏移。
* 向后移动必须小于当前后边的空闲空间大小，写入数据必须小于缓冲区剩余空间大小，不足就进行扩容！
确保可写空间足够 :
* 末尾空闲空间足够直接返回。
* 末尾空间不足，但算上起始位置的空闲空间大小足够，将数据移动到起始位置。
* 如果总空间不足，进行扩容，扩容到足够空间即可
写入数据：首先保证有足够空间，然后将数据数据拷贝进去。可以继续设计出针对string的写入、针对Buffer的写入以及写入 + 压入数据
读取数据：首先读取大小len必须小于可读取数据大小，然后拷贝数据出来。同样可以设计出针对string的读取、针对Buffer的读取以及读取+弹出数据。
清空缓冲区：将偏移量归零即可！
读取一行数据：先找到换行符，然后进行读取。

// 缓冲区Buffer类
class Buffer
{
private:
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;

private:
    char *Begin() { return &*_buffer.begin(); }

public:
    Buffer() : _buffer(BUFFER_DEFAULT_SIZE), _reader_idx(0), _writer_idx(0) {}
    // 获取当前写入起始地址
    char *WritePos() { return Begin() + _writer_idx; }
    // 获取当前读取起始地址
    char *ReadPos() { return Begin() + _reader_idx; }
    // 获取读取位置之前的空闲空间
    uint64_t HeadIdleSize() { return _reader_idx; }
    // 获取写入位置之后的空闲空间
    uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }
    // 获取可读数据大小
    uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }
    // 读/写偏移移动
    void MoveReadOffset(uint64_t len)
    {
        if (len <= 0)
            return;
        assert(len <= ReadAbleSize());
        _reader_idx += len;
    }
    void MoveWriteOffset(uint64_t len)
    {
        if (len <= 0)
            return;
        assert(len <= TailIdleSize());
        _writer_idx += len;
    }
    // 确保可写空间足够 --- 整体空闲空间足够了就移动数据，否则进行扩容
    void EnsureWriteSpace(uint64_t len)
    {
        // 当len小于写入位置之后的空闲空间 直接进行写入
        if (len <= TailIdleSize())
        {
            return;
        }

        // len 小于总共的空闲空间
        else if (len <= TailIdleSize() + HeadIdleSize())
        {
            // 记录总共的数据
            uint64_t sz = ReadAbleSize();
            // 移动数据
            std::copy(ReadPos(), ReadPos() + sz, Begin());
            // 更新写入读取位置
            _reader_idx = 0;
            _writer_idx = sz;
        }
        // 需要扩容
        else
        {
            // 在写入位置之后扩充len个大小
            _buffer.resize(_writer_idx + len);
        }
    }
    // ----------写入数据------------
    void Write(const void *data, uint64_t len)
    {
        // 写入的数据不能超过可写的总空间
        // 确保可以正常写入
        EnsureWriteSpace(len);
        // 进行拷贝
        const char *d = reinterpret_cast<const char *>(data);
        std::copy(d, d + len, WritePos());
    }
    // 写入Buffer
    void WriteBuffer(Buffer &buffer)
    {
        // 直接调用Write
        return Write(buffer.ReadPos(), buffer.ReadAbleSize());
    }
    // 写入字符串
    void WriteString(const std::string &str)
    {
        // 直接调用Write
        return Write(&str[0], str.size());
    }
    // 写入 + 偏移
    void WriteAndPush(const void *data, uint64_t len)
    {
        if (len <= 0)
            return;
        // 进行写入
        Write(data, len);
        // 更新写入偏移量
        MoveWriteOffset(len);
    }
    void WriteStringAndPush(const std::string &str)
    {

        // 直接调用Write
        WriteString(str);
        LOG(DEBUG, "%s
", WritePos());
        // 更新偏移量
        MoveWriteOffset(str.size());
        LOG(DEBUG, "当前可读数据大小：%ld
", ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer &buffer)
    {
        // 直接调用Write
        Write(buffer.ReadPos(), buffer.ReadAbleSize());
        MoveWriteOffset(buffer.ReadAbleSize());
    }

    //---------读取数据----------
    void Read(void *buf, uint64_t len)
    {
        // len 必须小于可读数据大小
        assert(len <= ReadAbleSize());
        // 进行拷贝
        std::copy(ReadPos(), ReadPos() + len, (char *)buf);
    }
    // 读取字符串
    std::string ReadAsString(uint64_t len)
    {
        // len 必须小于可读数据大小
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        // 直接调用Read
        Read(&str[0], len);
        return str;
    }
    // 读取 + Pop
    void ReadAndPop(void *buf, uint64_t len)
    {
        // 进行读取
        Read(buf, len);
        // 更新读取偏移量
        MoveReadOffset(len);
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        // len 必须小于可读数据大小
        assert(len <= ReadAbleSize());
        // 进行读取
        std::string str = ReadAsString(len);
        // 更新偏移量
        MoveReadOffset(len);
        return str;
    }

    //-----------读取一行数据--------------
    char *FindCRLF()
    {
        return (char *)memchr(ReadPos(), '
', ReadAbleSize());
    }
    std::string GetLine()
    {
        // 先寻找换行符
        char *pos = FindCRLF();
        if (pos == nullptr)
            return "";
        // 读取要带走'
'
        std::string str = ReadAsString(pos - ReadPos() + 1);
        return str;
    }
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }

    // 清除数据
    void Clear()
    {
        // 偏移量归零即可
        _reader_idx = 0;
        _writer_idx = 0;
    }
};

2.2 通用类型 Any类

每⼀个Connection对连接进⾏管理，最终都不可避免需要涉及到应⽤层协议的处理，因此在Connection中需要设置协议处理的上下⽂来控制处理节奏。但是应⽤层协议千千万，为了降低耦合度，这个协议接收解析上下文就不能有明显的协议倾向，它可以是任意协议的上下文信息，因此就需要⼀个通⽤的类型来保存各种不同的数据结构：

一个连接必须拥有一个请求接收与解析的上下文！
上下文的类型或者结构不能固定！因为服务器的协议支持的协议很多，不同的协议，可能都有不同的上下文结构！

所以必须拥有一个容器，能够保存各种不同的类型！那么就要实现一个any类

假如使用模版类方法，那么实例化对象的时候一定要指明容器保存的数据类型！而我们需要的是any可以接收任意类型Any a ; a = 10 ; a = "abc"!
但是可以嵌套一下，在Any类中设计一个类，专门用于保存各种类型的数据，而Any类保存的是固定类的对象。
对于这个固定类依旧不能使用模版。但这里这里可以采用多态，设计一个子类，这是一个模版类。
这样可以通过父类指针读取子类数据！
Any类中，保存的是holder类的指针，当Any类需要保存一个数据时，只需要提供placeholder子类实例化一个
特定类型的对象出来，让子类对象保存数据！

// 通用类型 Any类
class Any
{
private:
    class holder
    {
    public:
        holder() {}
        virtual ~holder() {};
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };
    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}

        virtual const std::type_info &type() override
        {
            return typeid(T);
        }
        virtual holder *clone() override
        {
            return new placeholder(_val);
        }

        T _val;
    };
    // 通过这个父类指针访问子类的成员
    holder *_content = nullptr;

public:
    Any()
    {
    }
    template <class T>
    // 拷贝构造
    Any(const T &val) : _content(new placeholder<T>(val))
    {
    }
    // 拷贝构造
    Any(const Any &other) : _content(other._content ? other._content->clone() : nullptr) {}
    // 交换数据
    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    // 赋值重载
    Any &operator=(Any &other)
    {
        // 根据 other建立一个临时对象 ，进行资源的交换
        Any(other).swap(*this);
        return *this;
    }
    template <class T>
    Any &operator=(const T &val)
    {
        // 根据 val 建立一个临时对象，进行资源的交换
        Any(val).swap(*this);
        return *this;
    }
    template <class T>
    T *Get()
    {
        if (typeid(T) != _content->type())
            return nullptr;
        return &((reinterpret_cast<placeholder<T> *>(_content))->_val);
    }

    ~Any()
    {
        delete _content;
    }
};

2.3 套接字 Socket模块

Socket模块是对套接字操作封装的⼀个模块，主要实现的socket的各项操作。是连接模块Connection 与监听模块Accpter的基础！对于套接字的操作我们已经在熟悉不过了：

构造函数析构函数
创建套接字
bool Create()
绑定地址信息
开始监听
向服务器发起连接
获取新连接
接收数据
发送数据
关闭套接字
创建一个服务端连接：首先创建套接字，然后将进程绑定地址信息，开启监听状态，注意设置为非阻塞读取（为了配合多路转接IO，不需要IO接口进行等待，一切等待都由多路转接负责）；同时启动地址重用，保护客户端
创建一个客户端连接：创建套接字，连接服务器。
设置套接字选项 — 开启地址端口重用
设置套接字阻塞属性 — 设置为非阻塞读取

// 套接字Socket类
class Socket
{
private:
    int _sockfd; // 套接字文件描述符

private:
    // 创建套接字
    bool Create()
    {
        // int socket(int domain, int type, int protocol);
        // IPV4     数据流IO
        _sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            LOG(ERROR, "create socket failed!
");
            return false;
        }
        LOG(INFO, "Sockfd:%d create success 
", _sockfd);
        return true;
    }
    // 绑定地址信息
    bool bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;                    // 使用IPv4版本地址
        addr.sin_port = htons(port);                  // 主机端口转网络端口
        addr.sin_addr.s_addr = inet_addr(ip.c_str()); // 将ip字符串转化为网络ip
        // 进行绑定
        socklen_t len = sizeof(struct sockaddr_in);
        int n = ::bind(_sockfd, (struct sockaddr *)&addr, len);
        if (n < 0)
        {
            LOG(ERROR, "bind failed!
");
            return false;
        }
        return true;
    }
    // 建立监听套接字
    bool Listen(int backlog = MAX_LIETENSIZE)
    {
        //                        全连接队列的大小
        int n = ::listen(_sockfd, backlog);
        if (n < 0)
        {
            LOG(ERROR, "listen failed!
");
            return false;
        }
        return true;
    }
    // 向服务器发起连接
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;                    // 使用IPv4版本地址
        addr.sin_port = htons(port);                  // 主机端口转网络端口
        addr.sin_addr.s_addr = inet_addr(ip.c_str()); // 将ip字符串转化为网络ip
        // 进行绑定
        socklen_t len = sizeof(struct sockaddr_in);
        int n = ::connect(_sockfd, (struct sockaddr *)&addr, len);
        if (n < 0)
        {
            LOG(ERROR, "connect failed!
");
            return false;
        }
        return true;
    }

    // 非阻塞启动地址重用
    void ReuseAddress(uint16_t port, const std::string &ip)
    {
        int val = 1;
        socklen_t len = sizeof(int);
        // 将端口号设置为可重用
        if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, len) == -1)
        {
            throw std::runtime_error("Failed to set SO_REUSEPORT");
        }

        // 将IP地址设置为可重用
        if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, len) == -1)
        {
            throw std::runtime_error("Failed to set SO_REUSEADDR");
        }
    }
    void NonBlock()
    {
        int flag = ::fcntl(_sockfd, F_GETFL, 0);
        ::fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }

public:
    // 构造函数
    Socket() : _sockfd(-1) {};
    Socket(int sockfd) : _sockfd(sockfd) {}
    // 析构函数
    ~Socket() { Close(); }
    // 返回套接字文件描述符
    int Sockfd() { return _sockfd; }

    // 接收数据
    ssize_t Recv(void *buf, size_t len, int flag = 0)
    {
        ssize_t n = ::recv(_sockfd, buf, len, flag);
        if (n <= 0)
        {
            // 被信号中断         没有数据了- 非阻塞读取完毕
            if (errno == EINTR || errno == EAGAIN)
                return 0; // 没有读取到数据

            LOG(ERROR, "recv failed!
");
            return -1;
        }
        return n; // 实际发送的长度
    }
    ssize_t NonBlockRecv(void *buf, size_t len)
    {
        //       MSG_DONTWAIT  非阻塞式读取
        if (len <= 0)
            return 0;
        return Recv(buf, len, MSG_DONTWAIT);
    }
    // 发送数据
    ssize_t Send(void *buf, size_t len, int flag = 0)
    {
        if (len <= 0)
            return 0;
        ssize_t n = ::send(_sockfd, buf, len, flag);
        if (n <= 0)
        {
            // 被信号中断         没有数据了- 非阻塞读取完毕
            if (errno == EINTR || errno == EAGAIN)
                return 0; // 没有读取到数据
            LOG(ERROR, "send failed!
");
            return -1;
        }
        LOG(DEBUG, "Send :%s
", (char *)buf);
        return n; // 实际发送的数据大小
    }
    ssize_t NonBlockSend(void *buf, size_t len)
    {
        //       MSG_DONTWAIT  非阻塞式发送
        if (len <= 0)
            return 0;
        return Send(buf, len, MSG_DONTWAIT);
    }

    // 创建服务端套接字
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flag = 0)
    {

        // 创建套接字
        if (Create() == false)
            return false;
        // 设置为非阻塞
        if (flag)
            NonBlock();
        // 将地址与端口设置为可重用
        ReuseAddress(port, ip);
        // 绑定地址信息
        if (bind(ip, port) == false)
            return false;
        // 进行监听
        if (Listen() == false)
            return false;

        return true;
    }
    // 创建客户端套接字
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        // 创建套接字
        if (Create() == false)
            return false;
        // 连接服务器
        if (Connect(ip, port) == false)
            return false;
        return true;
    }
    // 服务器获取连接
    int Accept()
    {
        int newfd = ::accept(_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            LOG(ERROR, "accept failed!
");
            return -1;
        }
        return newfd;
    }
    //  关闭套接字
    void Close()
    {
        LOG(INFO, "Close sockfd: %d
", _sockfd);
        if (_sockfd != -1)
        {
            ::close(_sockfd);
        }
    }
};