当前位置：首页 > news >正文

IO多路复用原理分析

news 2026/5/12 4:03:14

介绍

什么是IO多路复用？

IO多路复用是一种高效的IO处理技术，它允许单个线程同时监控多个文件描述符（如网络连接、文件、管道等），当其中任何一个准备好读写操作时，系统会立即通知程序进行处理。这种机制避免了传统阻塞IO中每个连接都需要独立线程等待的问题，极大提升了并发处理能力。具体来说，当程序调用select、poll或epoll等系统调用时，内核会同时检查所有被监控的文件描述符状态，只返回那些已经就绪的IO操作，这样程序就能在非阻塞的情况下高效处理成千上万个并发连接，是现代高并发服务器的核心技术基础。

传统阻塞IO的问题

// 每个连接需要一个线程funchandleConn(conn net.Conn){buf:=make([]byte,1024)// 这里会阻塞等待数据n,err:=conn.Read(buf)// 线程挂起，直到有数据// 处理数据...}

IO多路复用：一个线程监控所有连接

// 一个线程监控所有连接for{// 询问内核：哪些连接有数据了？readyConns:=poll()// 返回有数据的连接列表for_,conn:=rangereadyConns{gohandleData(conn)// 只处理有数据的连接}}

类型

select

在传统阻塞IO模型下，每个网络连接都需要独立线程阻塞等待数据，造成大量线程资源浪费；而select机制通过IO多路复用技术，实现了单线程对最多1024个连接的高效批量监控，将分散的阻塞等待转化为统一的事件驱动处理，显著提升了网络IO的并发处理能力和系统吞吐量。下面是一个select的使用示例

// 用户空间：告诉内核要监控哪些fdFD_SET(fd1,&readfds);// 监控fd1读FD_SET(fd2,&readfds);// 监控fd2读FD_SET(fd3,&writefds);// 监控fd3写// 进入内核：阻塞等待事件select(nfds,&readfds,&writefds,&exceptfds,timeout);// 返回后：内核告诉你哪些fd就绪了if(FD_ISSET(fd1,&readfds)){// fd1有数据了，可以读了}if(FD_ISSET(fd3,&writefds)){// fd3可以写了}

内核层发生了什么？

// 简化版内核逻辑intdo_select(intn,fd_set*readfds,fd_set*writefds,fd_set*exceptfds){while(1){intretval=0;// 遍历所有fd（0到n-1）for(inti=0;i<n;i++){structfile*file=fget(i);if(!file)continue;// 检查这个fd是否有读事件if(readfds&&FD_ISSET(i,readfds)){if(file->f_op->poll(file)&POLLIN){FD_SET(i,readfds);// 标记为就绪retval++;}}// 检查写事件...// 检查异常事件...}if(retval>0)returnretval;// 有fd就绪了if(timeout&&time_after(timeout))return0;// 超时schedule();// 让出CPU，睡眠等待}}

可以看出，内核层每次需要遍历所有的fd。为什么最多只能等待1024个连接？因为select使用bitmap来存储文件描述符，且这个长度由于历史包袱，被限制死1024了。不好改，所以情况就是这么个情况~

// POSIX标准规定：#defineFD_SETSIZE1024

#include<sys/select.h>#include<stdio.h>intmain(){fd_set readfds;// 尝试设置第1024个fdFD_SET(1023,&readfds);// ✅ 成功// 尝试设置第1025个fdFD_SET(1024,&readfds);// ❌ 数组越界！printf("FD_SETSIZE = %d\n",FD_SETSIZE);// 输出：1024return0;}

poll

poll机制针对select的1024连接限制进行了关键改进，摒弃了固定大小的位图结构，转而采用动态数组来存储文件描述符信息，这种设计不仅突破了连接数量的硬性上限，使其能够轻松支持成千上万的并发连接，还提供了更灵活的事件管理能力，为大规模网络应用奠定了基础架构优势。像下面这样

#include<poll.h>intmain(){intlisten_sock=socket(AF_INET,SOCK_STREAM,0);bind(listen_sock,...);listen(listen_sock,5);// poll用数组，不是位图structpollfdfds[2000];// ✅ 可以轻松超过1024intnfds=0;// 添加监听socketfds[0].fd=listen_sock;fds[0].events=POLLIN;// 监听读事件fds[0].revents=0;// 内核返回的事件nfds=1;while(1){// 调用pollintret=poll(fds,nfds,-1);if(ret>0){// 检查监听socketif(fds[0].revents&POLLIN){// 有新连接intnew_client=accept(listen_sock,...);// 添加到poll数组fds[nfds].fd=new_client;fds[nfds].events=POLLIN;fds[nfds].revents=0;nfds++;}// 检查客户端socketfor(inti=1;i<nfds;i++){if(fds[i].revents&POLLIN){// 有数据charbuffer[1024];intn=read(fds[i].fd,buffer,sizeof(buffer));if(n<=0){// 客户端断开，从数组移除close(fds[i].fd);// 用最后一个元素覆盖当前元素fds[i]=fds[nfds-1];nfds--;i--;// 重新检查当前位置}else{handle_data(buffer,n);}}}}}}

但是轮询所有的fd的问题，依然存在呢~~时间复杂度还是O(n)的

epoll

epoll在poll基础上实现了革命性优化，其命名中的"e"代表事件（event），彰显其事件驱动的核心设计理念。与poll的主动轮询机制不同，epoll采用被动事件通知模式，通过内核维护的就绪队列和回调机制，仅当文件描述符状态发生变化时才触发处理，这种设计不仅避免了无效遍历，更将时间复杂度从O(n)降至O(1)，使单线程能够高效管理数十万并发连接，真正实现了高性能的IO多路复用。下面是一段示例代码

// ep_poll - epoll的核心函数staticintep_poll(structeventpoll*ep,structepoll_event__user*events,intmaxevents,structtimespec*timeout){intres,avail;structepitem*epi;structepoll_event__user*eventpoll_buf;unsignedlongflags;// 关键：直接获取就绪队列！O(1)复杂度！mutex_lock(&ep->mtx);avail=ep->ovflist!=EP_UNACTIVE_PTR;if(!avail)avail=ep->rdllink.next!=&ep->rdllist;// 检查就绪队列是否为空mutex_unlock(&ep->mtx);if(avail){// 有就绪的fd，直接拷贝给用户空间res=ep_send_events(ep,events,maxevents);returnres;}// 没有就绪的fd，睡眠等待for(;;){set_current_state(TASK_INTERRUPTIBLE);// 再次检查就绪队列mutex_lock(&ep->mtx);avail=ep->ovflist!=EP_UNACTIVE_PTR;if(!avail)avail=!list_empty(&ep->rdllist);mutex_unlock(&ep->mtx);if(avail||timed_out)break;// 睡眠等待事件if(!schedule_hrtimeout_range(to,slack,HRTIMER_MODE_ABS))timed_out=1;}__set_current_state(TASK_RUNNING);returnep_send_events(ep,events,maxevents);}// ep_send_events - 发送就绪事件staticintep_send_events(structeventpoll*ep,structepoll_event__user*events,intmaxevents){structepitem*epi,*tmp;structepoll_eventevent;unsignedlongflags;intcnt=0;// 关键：只遍历就绪队列！不是遍历所有fd！list_for_each_entry_safe(epi,tmp,&ep->rdllist,rdllink){// 检查事件是否仍然有效if(ep_item_poll(epi,&pt)){// 拷贝到用户空间__put_user(epi->event.events,&event.events);__put_user(epi->event.data,&event.data);__copy_to_user(&events[cnt++],&event,sizeof(event));if(cnt>=maxevents)break;}}returncnt;}

// 每个epoll实例structeventpoll{spinlock_tlock;// 自旋锁structmutexmtx;// 互斥锁wait_queue_head_twq;// 等待队列structlist_headrdllist;// 关键：就绪fd链表！structrb_root_cachedrbr;// 红黑树，存储所有监听的fdstructepitem*ovflist;// 溢出链表};// 每个监听的fdstructepitem{union{structrb_noderbn;// 红黑树节点structrcu_headrcu;};structlist_headrdllink;// 就绪链表节点structepitem*next;// 溢出链表structepoll_filefdffd;// 文件描述符信息intnwait;// 等待队列数量structlist_headpwqlist;// 等待队列structeventpoll*ep;// 所属的epoll实例structlist_headfllink;// 文件链表structepoll_eventevent;// 事件类型};

下面是添加一个fd到epoll的例子。当fd就绪的时候，它会触发一个回调，添加到就绪队列里面，使用这种事件驱动的方式，把O(n)优化到了O(1)。这样就使得效率大大提升了

// ep_insert - 添加fd到epollstaticintep_insert(structeventpoll*ep,structepoll_event*event,structfile*tfile,intfd){structepitem*epi;structep_pqueueepq;// 分配epitemepi=kmem_cache_alloc(epi_cache,GFP_KERNEL);// 初始化epitemepi->ep=ep;epi->event=*event;epi->ffd.file=tfile;epi->ffd.fd=fd;// 添加到红黑树（O(log n)）ep_rbtree_insert(ep,epi);// 设置回调函数 - 关键！epq.epi=epi;init_poll_funcptr(&epq.pt,ep_ptable_queue_proc);// 调用文件的poll函数，注册回调revents=tfile->f_op->poll(tfile,&epq.pt);// 如果已经有事件，直接添加到就绪队列if(revents&event->events){list_add_tail(&epi->rdllink,&ep->rdllist);}return0;}// 回调函数 - fd就绪时被调用staticvoidep_ptable_queue_proc(structfile*file,wait_queue_head_t*whead,poll_table*pt){structep_pqueue*epq=container_of(pt,structep_pqueue,pt);structepitem*epi=epq->epi;// 添加到等待队列，设置回调add_wait_queue(whead,&epi->wait);}