004 仿muduo实现高性能服务器组件_Buffer模块与Socket模块的实现

​🌈个人主页：Fan_558
🔥 系列专栏：仿muduo
🌹关注我💪🏻带你学更多知识

前言

这章将会向你介绍仿muduo高性能服务器组件的buffer模块与socket模块的实现

Buffer模块

设计思想

实现思想：

1、实现缓冲区得有一块内存空间，采用vector，string字符串的操作遇到’\0’就停止了，网络操作中什么样的数据都有，'\0’可能也有，string大部分的操作都是字符串操作，所以不太行

2、记录当前的读取数据位置与当前的写入数据位置，避免每次写入数据需要重新遍历数组找写入读入位置

3、考虑整体缓冲区空闲空间是否足够 （因为读位置也会向后偏移，前边有可能会有空间） 足够：则将数据（读位置开始）移动到起始位置即可
不够：扩容，从当前写位置开始扩容足够大小 数据一旦写入成功，当前写位置就要向后偏移

4、读取数据/写入数据
当前的读取/写入位置指向哪里，就从哪里开始读取/写入，前提是有数据可读/有空间可写，读取/写入完数据，读偏移/写偏移向后偏移

为了方便查阅

代码如下：

class Buffer{
private:
    std::vector<char> _buffer; //使用vector进行内存空间管理
    uint64_t _reader_idx; //读偏移
    uint64_t _writer_idx; //写偏移
public:
    Buffer():_reader_idx(0), _writer_idx(0) ,_buffer(BUFFER_SIZE) {}
    //获取_buffer起始元素的地址
    char* begin() {return &*_buffer.begin();}
    //获取当前写入起始地址（_buffer的空间起始地址，加上写偏移量
    char* WritePos() { return begin() + _writer_idx; }
    //获取当前读取起始地址（_buffer的空间起始地址，加上读偏移量
    char* ReadPos() { return begin() + _reader_idx; }
    //获取缓冲区末尾空闲空闲大小--写偏移之后的空闲空间
    uint64_t TailIdleSize() {return _buffer.size() - _writer_idx; }
    //获取缓冲区起始地址空闲空间大小--读偏移之前的空闲空间
    uint64_t HeadIdleSize() {return _reader_idx; }
    //获取可读数据大小
    uint64_t ReadAbleSize() {return _writer_idx - _reader_idx; }


    //读取数据后，将读偏移向后移动
    void MoveReadOffest(uint64_t len) 
    {   
        //向后移动的大小，必须小于可读数据大小
        assert(len <= ReadAbleSize());
        _reader_idx += len; 
    }
    //写入数据后，将写偏移向后移动
    void MoveWriteOffest(uint64_t len) { _writer_idx += len; }
    //确保可写空间足够（整体空闲空间够了就移动数据，否则就扩容）
    void EnsureWriteSpace(uint64_t len)
    {
        //如果末尾空闲空间大小足够，直接返回
        if(len < TailIdleSize()) return;
        //如果不够，判断加上起始位置的空闲空间大小是否足够，够了就将可读数据移动到起始位置
        else if(len <= HeadIdleSize() + TailIdleSize()) 
        {
            uint64_t sz = ReadAbleSize();   //可读数据大小
            _reader_idx = 0;    //更新读偏移
            _writer_idx = sz;   //更新写偏移
            return;
        }
        //总体空间不够，则需要扩容，不移动数据，直接给写偏移之后扩容足够空间即可
        else _buffer.resize(_writer_idx + len);
    }
    //写入数据
    void Write(const void* data, uint64_t len)
    {
        //保证是否有足够空间
        EnsureWriteSpace(len);
        const char* d = (const char* )data;
        //拷贝数据到buffer当中
        std::copy(d, d + len, WritePos());
    }
    void WriteAndPush(const void* data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffest(len);
    }
    //写入一个字符串
    void WriteString(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }
    //向buffer中写入一个字符串并向后移动write
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriteOffest(data.size());
    }
    //把一个buffer类型的数据写入
     void WriteBuffer(Buffer &data)
     {
        Write(data.ReadPos(), data.ReadAbleSize());
     }
    //向buffer中写入一个并向后移动write
     void WriteBufferAndPush(Buffer &data)
     {
        WriteBuffer(data);
        MoveWriteOffest(data.ReadAbleSize());
     }
    //读取数据
    void Read(void* buf, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        //保持参数类型一致
        std::copy(ReadPos(), ReadPos() + len, (char*)buf);
    }
    void ReadAndPop(void* buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffest(len);
    }
    //把读取的数据当作一个string返回  
    std::string ReadAsString (uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        //从缓冲区中读取长度为len的数据，并将其存储到字符串str的内存地址开始处的位置
        Read(&str[0], len);
        return str;
    }
    //读取一个string并向后移动（确保下一次不会重复读取）
    std::string ReadAsStringAndPop(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffest(len);
        return str;
    }
    /*由于后面我们的高并发服务器会支持应用层协议的HTTP，而在HTTP协议中通常就是读取一行的数据，因为
    请求行和请求报头以及响应行和响应报头都是以\r\n作为分隔符的，都是一行行的数据
    所以我们的缓冲区也提供一个查找换行字符的位置*/
    char* FindCRLF()
    {
        //在可读数据范围内查找第一个出现的换行符的位置
        char* res = (char*)memchr(ReadPos(), '\n', ReadAbleSize());
        return res;
    }
    //获取一行数据
    std::string Getline()
    {
        char* pos = FindCRLF();
        if(pos == nullptr) return "";
        /*将换行符\n前的数据读出，+1：包括换行符（不然的话下一次再查找，换行符就在开头） */
        return ReadAsString(pos - ReadPos() + 1); 
       
    }
    //读出一行数据后，将读偏移向后移
    std::string GetLineAndPop()
    {
        std::string str = Getline();
        MoveReadOffest(str.size());
        return str;
    }
    //清空缓冲区
    void clear()
    {
        //只需要将偏移量归零
        _writer_idx =  _reader_idx = 0;
    }
};

Socket模块

设计思想：

在该模块当中除了对socket套接字原有的操作进行封装，还提供了直接创建服务端和客户端连接的接口
为了方便查阅

代码如下

#define MAX_LISTEN 1024
class Socket{
    private:
    int _sockfd;
    public:
        Socket()
        :_sockfd(-1)
        {}
        Socket(int fd)
        :_sockfd(fd)
        {}
        //关闭套接字
        ~Socket() { Close(); }
        int Fd()
        {
            return _sockfd;
        }
        //创建套接字
        bool Create()
        {
            //int socket(int domain, int type, int protocol)  AF_INET: 表示使用ipv4地址族 SOCK_STREM: 表示创建面向连接的套接字类(TCP) IPPROTO_TCP: 表示使用TCP协议
            _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
            if(_sockfd < 0)
            {
                ERR_LOG("CREATE SOCKET FAILEDQ!");
                return false;
            }
            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地址转化为网络字节序的32位ipv4地址
            socklen_t len = sizeof(struct sockaddr_in);
            //int bind(int socket, const struct sockaddr *addr. socklen_t addrlen);
            int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("BIND ADDRESS FAILEDQ!");
                return false;
            }
            return true;
        }
        //开始监听
        bool Listen(int backlog = MAX_LISTEN)
        {
            int ret = listen(_sockfd, backlog);
            if(ret < 0){
                ERR_LOG("SOCKET LISTEN FAILED!");
                return false;
            }
            return true;
        }
        //向服务器发起连接(传入服务器的ip和端口信息)
        bool Connect(const std::string &ip, uint16_t port)
        {
            //int connect(int sockfd, const struct sockaddr* addr, socklen_t addrlen);
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("CONNECT SERVER FAILEDQ!");
                return false;
            }
            return true;
        }
        //监听有新连接后，获取新连接（返回一个文件描述符）
        int Accept()    
        {
            int newfd = accept(_sockfd, nullptr, nullptr);
            if(newfd < 0)
            {
                ERR_LOG("SOCKET ACCEPT FAILED!");
                return -1;
            }
            return newfd;
        }
        //接收数据(ssize_t为有符号整数，size_t无符号整数，默认0为阻塞操作)
        ssize_t Recv(void* buf, size_t len, int flag = 0)
        {
            ssize_t ret = recv(_sockfd, buf, len, flag);
            if(ret <= 0)
            {
                //EAGAIN 当前socket的接收缓冲区中没有数据了，在非阻塞的情况下才会有这个错误
                //EINTR 当前socket的阻塞等待被信号打断了
                if(errno == EAGAIN || errno == EINTR)
                    return 0;
                else
                {
                    ERR_LOG("SOCKET RECV FAILED");
                    return -1;
                }
            }
            return ret; //返回实际接收的数据长度
        }
        ssize_t NonBlockRecv(void* buf, size_t len)
        {
            return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞
        }
        //发送数据
        ssize_t Send(const void* buf, size_t len, int flag = 0)
        {
            ssize_t ret = send(_sockfd, buf, len, flag);
            if(ret < 0)
            {
                if(errno == EAGAIN || errno == EINTR)
                {
                return 0;
                }
                ERR_LOG("SOCKET RECV FAILED");
                return -1;
            }
            return ret; //返回实际发送的数据长度
        }
        ssize_t NonBlockSend(void* buf, size_t len)
        {
            Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞
        }
        //关闭套接字
        void Close()
        {
            if(_sockfd != -1){
                close(_sockfd);
                _sockfd = -1;
            }
        }
        //创建一个服务端连接
        bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false)
        {
            if(Create()==false) return false;
            //是否启动非阻塞
            if(block_flag) NonBlock();
            if(Bind(ip, port) == false) return false;
            if(Listen() == false) return false;
            ReuseAddress();
            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;
        }
        //设置套接字选项---开启地址端口重用
        void ReuseAddress()
        {
            // int setsockopt(int fd, int leve, int optname, void *val, int vallen)
            int val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
            val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
        }
        //设置套接字阻塞属性---设置为非阻塞
        void NonBlock()
        {
            //int fcntl(int fd, int cmd, ... /* arg */ );
            int flag = fcntl(_sockfd, F_GETFL, 0);
            fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
        }
};

小结

今日的项目分享就到这里啦，下一篇将会向你介绍Channel与Poller模块