【Linux】多路转接 Select ， Poll和Epoll

打不了嗝 ᥬ᭄

3794人浏览 · 2025-12-06 16:29:22

打不了嗝 ᥬ᭄ · 2025-12-06 16:29:22 发布

一. Select

1. select 函数原型

select 函数是I/O多路复用的经典实现，其基本原型如下：

int select（int nfds，fd_set *readfds，fd_set *writefds，fd_set *exceptfds，struct timeval *timeout）

参数

nfds：记录监控最大的文件描述符加1，得到这个参数。用于内部循环数组找到最大文件描述符。

readfds，writefds，exceptfds：这三个参数分别代表读，写和异常监视的文件描述符集合。使用的是fd_set类型表示，用位图来表示文件描述符是否启用。

{

FD_SET(fd，&set)：将文件描述符fd添加到集合set中

FD_CLR(fd，&set)：从集合set移除文件描述符fd

FD_ISSET(fd，&set)：检查文件描述符fd是否被加入集合set

FD_ZERO(&set)：清空集合set中的所有文件描述符 }

timeout：这是一个指向timeval结构的指针，该结构用于设定select等待 I/O 事件的超时时间。

struct timeval {

long tv_sec； //

long tv_usec；

}

若 timeval 为nullptr，那么就永久阻塞直到有文件描述符就绪

若{0,0} ，非阻塞立即返回

若{5,0}，非阻塞间隔5秒后返回

返回值：

大于0：就绪的文件描述符数量

等于0：超时

小于0：发生错误

2. select 执行过程

用户用FD_ZERO或FD_SET初始化fd_set，标记需要监听的描述符。fd_set 本质是数组，存储被监听的文件描述符fd。接着调用select进入内核态。内核通过参数遍历数组检查文件对应是否有就绪事件。

3. select 的特点

select 每次都需要遍历一遍数组查看事件是否就绪，一旦文件描述符增多这样效率就会降低。而且存在着描述符数量上限的限制。当事件超出文件描述符上限时就无法进行轮询查看。而且select存在着文件描述符上限，因为它是用位图存储文件描述符，意味着最多能容纳1024

4. select server 实现

#pragma once

#include "Log.hpp"
#include "InetAddr.hpp"
#include "Socket.hpp"
#include <memory>
#include <iostream>
#include <unistd.h>
using namespace std;
using namespace LogModule;
using namespace SocketModule;

class SelectServer
{
    const static int size = sizeof(fd_set) * 8;
    const static int defaultfd = -1;

public:
    SelectServer(uint16_t port)
        : _listensock(make_unique<TcpSocket>()),
          _running(false)
    {
        _listensock->BuildTcpSocketMethod(port);
        for (int i = 0; i < size; i++)
            _fd_array[i] = defaultfd;
        _fd_array[0] = _listensock->Fd();
    }

    void Start()
    {
        _running = true;
        while (_running)
        {
            struct timeval timeout = {3,0};
            fd_set rfds;
            FD_ZERO(&rfds);
            int maxfd = defaultfd;
            for (int i = 0; i < size; i++)
            {
                if (_fd_array[i] == defaultfd)
                    continue;

                FD_SET(_fd_array[i], &rfds); // 重置rfds
                if (_fd_array[i] > maxfd)
                {
                    maxfd = _fd_array[i];
                }
            }

            Print();

            int n = select(maxfd + 1, &rfds, nullptr, nullptr, &timeout);
            // 检测时要比maxfd大1 ，读，写，错误，timeout（是否阻塞，隔多长时间检测一次）
            switch (n)
            {
            case 0:
                LOG(LogLevel::INFO) << "timeout.....";
                break;
            case -1:
                LOG(LogLevel::ERROR) << "Select 错误";
                break;
            default:
                LOG(LogLevel::INFO) << "事件准备就绪......";
                Dispachar(rfds);
                break;
            }
        }
        _running = false;
    }

    void Dispachar(fd_set &rfds) // 连接器
    {
        for (int i = 0; i < size; i++)
        {
            if (_fd_array[i] == defaultfd)
                continue;

            if (FD_ISSET(_fd_array[i], &rfds))
            {
                if (_fd_array[i] == _listensock->Fd())
                {
                    Accepter(); // 如果是第一次加入就设置进入 _fd_array
                }
                else
                {
                    Recver(_fd_array[i],i); // 如果不是第一次就执行IO流
                }
            }
        }
    }

    void Accepter()
    {
        InetAddr client;
        int sockfd = _listensock->Accept(&client);

        if (sockfd > 0)
        {
            LOG(LogLevel::INFO)<<"sockfd success,port:"<<client.Port()<<"  Addr:"<<client.StringAddr();

            int pos = 0;
            for (; pos < size; pos++)
            {
                if (_fd_array[pos] == defaultfd)
                    break;
            }
            if (pos == size)
            {
                LOG(LogLevel::ERROR) << "select full";
                close(sockfd);
            }
            else
            {
                _fd_array[pos] = sockfd;
            }
        }
    }

    void Recver(int fd,int pos)
    {
        char buffer[1024];
        ssize_t n = recv(fd,buffer,sizeof(buffer)-1,0);
        if(n>0)
        {
            buffer[n] = 0;
            cout<<"client say#"<<buffer<<endl;
        }
        else if(n == 0)
        {
            LOG(LogLevel::INFO)<<"client quit";
            _fd_array[pos] = defaultfd;
            close(fd);
        }
        else
        {
            LOG(LogLevel::ERROR)<<"recv error";
            _fd_array[pos] = defaultfd;
            close(fd);
        }
    }

    void Stop()
    {
        _running = false;
    }

    void Print()
    {
        cout << "_fd_arrar[]:";
        for (int i = 0; i < size; i++)
        {
            if (_fd_array[i] == defaultfd)
                continue;
            cout << _fd_array[i];
        }
        cout << endl;
    }

    ~SelectServer()
    {
    }

private:
    unique_ptr<Socket> _listensock;
    int _fd_array[size];
    bool _running;
};

二. Poll

1. poll 函数原型

poll 函数基本原型如下：

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

参数：

fds：是一个结构体数组指针，每个元素对应一个要监听的描述符

{ fd // 该事件的fd

revents // 实际发生的事件（输出）

events // 想要监听的事件（输入）

}

// POLLIN ：请求监听事件可读

// POLLOUT：请求监听事件可写

// POLLPRI：请求监听紧急数据可读事件

nfds：要监听的描述符数量

timeout：超时时间，-1为永久阻塞，0为非阻塞，大于0阻塞时长

2. poll 执行过程

准备调用前先定义结构体数组 pollfd ，为每个要监听的描述符配置参数。fd 指定要监听的文件描述符，events 设置监听事件（可读可写），revents初始化为0。接着确定要监听的描述符数量 nfds，设置超时时间。接着调用 poll 进入内核态。遍历数组结构体，若事件就绪记录事件执行。返回就绪描述符。用户态恢复处理就绪事件

3. poll 的优缺点

无文件描述符限制，摆脱了位图采用动态数组进行管理。同时将事件与描述符绑定，输入输出绑定。不需要分批管理。但同时它也没有摆脱需要进行整体遍历的缺点，一旦事件增多相应的效率也下降。从用户态到内核态拷贝开销大。每次调用都要将poll从用户态拷贝到内核态。

4. poll server 实现

#pragma once

#include <iostream>
#include "Log.hpp"
#include "InetAddr.hpp"
#include <unistd.h>
#include "Socket.hpp"
#include <sys/poll.h>
#include <memory>

using namespace std;
using namespace SocketModule;
using namespace LogModule;

class PollServer
{
    const static int size = 4096;
    const static int defaultfd = -1;

public:
    PollServer(uint16_t port)
        : _listensock(make_unique<TcpSocket>()),
          _running(false)
    {
        _listensock->BuildTcpSocketMethod(port);
        for (int i = 0; i < size; i++)
        {
            _fd_array[i].fd = defaultfd;
            _fd_array[i].events = 0;
            _fd_array[i].revents = 0;
        }
        _fd_array[0].fd = _listensock->Fd();
        _fd_array[0].events = POLLIN;
    }

    void Start()
    {
        int tiemout = -1;
        _running = true;
        while (_running)
        {
            int n = poll(_fd_array, size, tiemout);
            switch (n)
            {
            case 0:
                LOG(LogLevel::INFO) << "timeout....";
                break;
            case -1:
                LOG(LogLevel::ERROR) << "poll error";
                break;
            default:
                LOG(LogLevel::INFO) << "事件就绪.....";
                Dispacher();
                break;
            }
        }
        _running = false;
    }

    void Dispacher()
    {
        for (int i = 0; i < size; i++)
        {
            if (_fd_array[i].fd == defaultfd)
                continue;
            if (_fd_array[i].revents && POLLIN)
            {
                if (_fd_array[i].fd == _listensock->Fd())
                {
                    Accepter();
                }
                else
                {
                    Recver(i);
                }
            }
        }
    }

    void Accepter()
    {
        InetAddr client;
        int sockfd = _listensock->Accept(&client);
        LOG(LogLevel::INFO) << "Get a new link" << sockfd << ",client is:" << client.StringAddr();
        if (sockfd >= 0)
        {
            int pos = 0;
            for (; pos < size; pos++)
            {
                if (_fd_array[pos].fd == defaultfd)
                    break;
            }
            if (pos == size)
            {
                LOG(LogLevel::DEBUG) << "poll full";
                close(sockfd);
            }
            else
            {
                _fd_array[pos].fd = sockfd;
                _fd_array[pos].events = POLLIN;
                _fd_array[pos].revents = 0;
            }
        }
    }

    void Recver(int pos)
    {
        char buffer[1024];
        ssize_t n = recv(_fd_array[pos].fd, buffer, sizeof(buffer) - 1, 0);
        if (n > 0)
        {
            buffer[n] = 0;
            cout << "client say#" << buffer << endl;
        }
        else if (n == 0)
        {
            LOG(LogLevel::INFO) << "client quit";
            close(_fd_array[pos].fd);
            _fd_array[pos].fd = defaultfd;
            _fd_array[pos].revents = 0;
            _fd_array[pos].events = 0;
        }
        else
        {
            LOG(LogLevel::ERROR) << "recv error";
            close(_fd_array[pos].fd);
            _fd_array[pos].fd = defaultfd;
            _fd_array[pos].revents = 0;
            _fd_array[pos].events = 0;
        }
    }

    void PrintFd()
    {
        cout << "_fd_array[]:";
        for (int i = 0; i < size; i++)
        {
            if (_fd_array[i].fd == defaultfd)
                continue;
            cout << _fd_array[i].fd << "  ";
        }
        cout << endl;
    }

    void Stop()
    {
        _running = false;
    }

    ~PollServer()
    {
    }

private:
    struct pollfd _fd_array[size];
    bool _running;
    unique_ptr<Socket> _listensock;
};

三. Epoll

1. epoll 函数原型

epoll 属于一个模型为的是解决 select和poll的痛点。一方面解决select文件描述符数量限制，另一方面解决poll遍历效率低的问题。

epoll_create：

int epoll_create(int size);
参数：

size：已废弃，只需要传大于0的数即可

返回值：

成功返回 epoll 描述符 epfd。失败返回-1

epoll_ctl：

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
参数：

epfd：epoll 描述符，由epoll_create 返回

op：操作符，对 epoll 模型的fd进行增加删除修改

{EPOLL_CTL_ADD：添加fd

EPOLL_CTL_DEL：删除fd

EPOLL_CTL_MOD：修改fd

}

fd：需要操作的fd

event：指定要监听的事件类型

{EPOLLIN：fd可读

EPOLLOUT：fd可写

EPOLLET：设置为ET模式

}

返回值：

成功0，失败-1

epoll_wait：

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
参数：

epfd：epoll 描述符，由epoll_create 返回

event：指定要监听的事件类型

{EPOLLIN：fd可读

EPOLLOUT：fd可写

EPOLLET：设置为ET模式

}

maxevents：最多处理的就绪fd数量

timeout：超时时间（-1永久阻塞，0立即返回不阻塞，大于0阻塞指定毫秒数）

返回值：

成功返回0，失败返回-1

2. epoll 执行过程

epoll 在 poll 的基础上提升了查找 fd 的效率，原因在于将 poll 储存fd用结构体数组，而 epoll 使用的是红黑树存储 fd。在此基础上，将就绪的 fd 放入就绪队列当中，这样就构成了一个完整的 epoll 模型。

首先创建 epoll 模型，epoll_create 构建出红黑树和就绪队列。接着进行epoll_ctl，本质是对红黑树进行增删改。最后等待fd事件在就绪队列中，epoll_wait进行等待事件发生。

对于就绪队列中的fd事件如何处理，epoll模型给出了两种模式选择

水平触发LT：epoll 模型默认的处理方式，当 epoll 检测到事件就绪时，先进行一部分处理，此时 epoll_wait 不会立即返回，而是等待缓冲区将数据全部处理完成后再进行返回。例如缓冲区有2K数据，第一次处理了1K，此时会等待读取另外1K后再进行返回

边缘触发ET：与水平触发相对，当缓冲区存在2K数据时，第一次处理了1K，此时epoll_wait不会等待而是直接进行返回。也就是说ET只有一次处理机会。

在 ET 模式下必须搭配非阻塞式的文件描述符使用。若当前的文件描述符是阻塞式的，ET在第一次接收缓冲区数据后进行挂起。阻塞式需要等待缓冲区被处理完成后才会向客户端进行应答。而ET模式下在提取一次缓冲区后需要等待下一次客户端发起请求后才会再次提取缓冲区。这也就导致了永久阻塞。所以说，ET模式下必须搭配非阻塞式文件描述符使用。

3. epoll 优缺点

首先，采用红黑树存储fd解决了文件描述符上限的限制，同时提升了查找效率。其次降低开销，不需要每次将fd集合拷贝到内核，而是通过mmap的方式进行内存映射。

4. epoll server 实现

#pragma once

#include <unistd.h>
#include <iostream>
#include <memory>
#include <sys/epoll.h>
#include "Log.hpp"
#include "InetAddr.hpp"
#include "Socket.hpp"

using namespace std;
using namespace LogModule;
using namespace SocketModule;

class EpollServer
{
    const static int size = 64;
    const static int defaultfd = -1;

public:
    EpollServer(uint16_t port)
        : _listensock(make_unique<TcpSocket>()),
          _isrunning(false),
          _epfd(defaultfd)
    {
        _listensock->BuildTcpSocketMethod(port);
        _epfd = epoll_create(220);
        if (_epfd < 0)
        {
            LOG(LogLevel::ERROR) << "epoll_create error";
            exit(EPOLL_CREATE_ERR);
        }
        LOG(LogLevel::INFO) << "epoll_create success:" << _epfd;

        struct epoll_event ev;
        ev.events = EPOLLIN;
        ev.data.fd = _listensock->Fd();

        int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock->Fd(), &ev);
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "epoll_ctl error";
            exit(EPOLL_CTL_ERR);
        }
    }

    void Start()
    {
        _isrunning = true;
        int timeout = -1;
        while (_isrunning)
        {
            int n = epoll_wait(_epfd, _revs, size, timeout);
            switch (n)
            {
            case -1:
                LOG(LogLevel::ERROR) << "epoll_wait error";
                break;
            case 0:
                LOG(LogLevel::INFO) << "timeout....";
                break;
            default:
                Dispacher(n);
                break;
            }
        }
    }

    void Dispacher(int num)
    {
        for (int i = 0; i < num; i++)
        {
            int sockfd = _revs[i].data.fd;
            uint32_t event = _revs[i].events;
            if (event & EPOLLIN)
            {
                if (sockfd == _listensock->Fd())
                {
                    Accepter();
                }
                else
                {
                    Recver(sockfd);
                }
            }
        }
    }

    void Accepter()
    {
        InetAddr client;
        int sockfd = _listensock->Accept(&client);
        if (sockfd >= 0)
        {
            LOG(LogLevel::INFO) << "有新链接链接,sockfd:" << sockfd << ",client ip:" << client.StringAddr();

            struct epoll_event ev;
            ev.events = EPOLLIN;
            ev.data.fd = sockfd;
            int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, sockfd, &ev);
            if (n < 0)
            {
                LOG(LogLevel::ERROR) << "epoll_ctl error";
            }
            else
            {
                LOG(LogLevel::INFO) << "新链接链接成功";
            }
        }
    }

    void Recver(int sockfd)
    {
        char buffer[1024];
        ssize_t n = recv(sockfd, buffer, sizeof(buffer) - 1, 0);
        if (n > 0)
        {
            buffer[n] = 0;
            cout << "client say# " << buffer;
        }
        else if (n == 0)
        {
            LOG(LogLevel::INFO) << "client quit";
            int ret = epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr);
            if (ret > 0)
            {
                LOG(LogLevel::INFO) << "成功退出sockfd";
            }
            close(sockfd);
        }
        else
        {
            LOG(LogLevel::DEBUG) << "recv error";
            int ret = epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr);
            if (ret > 0)
            {
                LOG(LogLevel::INFO) << "成功退出sockfd";
            }
            close(sockfd);
        }
    }

    void Stop()
    {
        _isrunning = false;
    }
    ~EpollServer()
    {
    }

private:
    unique_ptr<Socket> _listensock;
    bool _isrunning;
    struct epoll_event _revs[size];
    int _epfd;
};