本文共 37202 字，大约阅读时间需要 124 分钟。
socket模型：
1、阻塞模型一个单进程accept阻塞，接收到客户端请求后，read消息，处理write返回，然后循环继续accept。
这种模型最最简单，不实际，没什么实际用途，对于新手教学还行。
2、多进程(线程)模型
主进程循环accept阻塞，接收到客户端请求后，fork子进程处理，子进程read阻塞，接收客户端消息并响应。
这种模型是我使用到最多的，简单实用，但是当客户端请求超多时，fork子进程多，系统资源消耗大，效果不理想；当然这种与多线程同理。
3、进程池(线程池)
主进程产生固定多的子进程，并定时监控子进程状态，初始子进程都为空闲状态。子进程在accept到客户端请求，通知主进程我很忙，然后处理请求，请求处理完成后，通知主进程我很闲。主进程主要监控子进程是否僵死或退出，维护进程池固定数量的进程来处理消息。
这种模型，可能每个人的实现方式不一样，这是我接触到的。优点是：不会产生超多的进程(线程)以至于过多消耗资源，在请求数量不多的情况下，效果还好；缺点是：因为是‘池’都有限制，当远远超过进程池限制的进程数，效果并不理想。
4、连接池
这种实现方式我第一次听说，在网上查了好久也没有头绪。在我的理解里，socket都是客户端向服务端发请求建立socket连接，因为客户端不同这种连接怎么重用？请高手指点一二，主要讲清原理即可。
5、select事件模型
这种实现方式是主进程将socket监听连接和client请求连接一起FD_SET到一个内核队列中，内核一直检查这个队列的哪个socket描述符有读或写或异常的响应则通知用户进程。用户进程检测到socket监听连接有响应，则accept与客户端建立连接，并把新的client请求连接FD_SET到内核队列中；如果检测到client请求连接有响应，则fork子进程，read客户端消息，处理并响应消息。
这种模型，select会捕获到你设置的某个socket描述符有可读可写或异常的事件，但是程序员需要自己检查自己设置的所有描述符，以确认是哪个描述符有事件发生。优点：占用资源少，不会消耗太多的cpu；缺点是select的效率和FD_SET到内核队列中的描述符的个数有关，当需要检测的描述符过多时，就要花费过多的时间去检测所有的描述符是否有时间发生，而且可以FD_SET的描述符内核也有限制，当客户端请求成千上万时，select便无能为力。
6、epoll事件模型
这种模型我刚开始接触，目前还没有完全使用或练习过。但是实现方式却是目前最好的一种模型，对设置监测的socket描述符同时设置回调函数，当内核监测到socket描述符有事件发生，则会主动触发回调函数。
这种模型，优点：可以接收任意多的连接，高效率，低消耗，稳定易使用；缺点：因为各系统提供的接口有很大的差异，可移植性差。
什么是epoll
epoll是什么？按照man手册的说法：是为处理大批量句柄而作了改进的poll。当然，这不是2.6内核才有的，它是在2.5.44内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44)，它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。
转自：http://blog.csdn.net/xiajun07061225/article/details/9250579
epoll的相关系统调用
epoll只有epoll_create,epoll_ctl,epoll_wait 3个系统调用。
1. int epoll_create(int size);
创建一个epoll的句柄。自从linux2.6.8之后，size参数是被忽略的。需要注意的是，当创建好epoll句柄后，它就是会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。
2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
epoll的事件注册函数，它不同于select()是在监听事件时告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型。
第一个参数是epoll_create()的返回值。
第二个参数表示动作，用三个宏来表示：
EPOLL_CTL_ADD：注册新的fd到epfd中；
EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
EPOLL_CTL_DEL：从epfd中删除一个fd；
第三个参数是需要监听的fd。
第四个参数是告诉内核需要监听什么事，struct epoll_event结构如下：
//保存触发事件的某个文件描述符相关的数据（与具体使用方式有关） typedef union epoll_data { void *ptr; int fd; __uint32_t u32; __uint64_t u64; } epoll_data_t; //感兴趣的事件和被触发的事件 struct epoll_event { __uint32_t events; /* Epoll events */ epoll_data_t data; /* User data variable */ };
events可以是以下几个宏的集合：
EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
EPOLLOUT：表示对应的文件描述符可以写；
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
EPOLLERR：表示对应的文件描述符发生错误；
EPOLLHUP：表示对应的文件描述符被挂断；
EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里
3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
收集在epoll监控的事件中已经发送的事件。参数events是分配好的epoll_event结构体数组，epoll将会把发生的事件赋值到events数组中（events不可以是空指针，内核只负责把数据复制到这个events数组中，不会去帮助我们在用户态中分配内存）。maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，如返回0表示已超时。
epoll工作原理
epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。
另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。
Epoll的2种工作方式-水平触发（LT）和边缘触发（ET）
假如有这样一个例子：
1. 我们已经把一个用来从管道中读取数据的文件句柄(RFD)添加到epoll描述符
2. 这个时候从管道的另一端被写入了2KB的数据
3. 调用epoll_wait(2)，并且它会返回RFD，说明它已经准备好读取操作
4. 然后我们读取了1KB的数据
5. 调用epoll_wait(2)......
Edge Triggered 工作模式：
如果我们在第1步将RFD添加到epoll描述符的时候使用了EPOLLET标志，那么在第5步调用epoll_wait(2)之后将有可能会挂起，因为剩余的数据还存在于文件的输入缓冲区内，而且数据发出端还在等待一个针对已经发出数据的反馈信息。只有在监视的文件句柄上发生了某个事件的时候 ET 工作模式才会汇报事件。因此在第5步的时候，调用者可能会放弃等待仍在存在于文件输入缓冲区内的剩余数据。在上面的例子中，会有一个事件产生在RFD句柄上，因为在第2步执行了一个写操作，然后，事件将会在第3步被销毁。因为第4步的读取操作没有读空文件输入缓冲区内的数据，因此我们在第5步调用 epoll_wait(2)完成后，是否挂起是不确定的。epoll工作在ET模式的时候，必须使用非阻塞套接口，以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。最好以下面的方式调用ET模式的epoll接口，在后面会介绍避免可能的缺陷。
i 基于非阻塞文件句柄
ii 只有当read(2)或者write(2)返回EAGAIN时才需要挂起，等待。但这并不是说每次read()时都需要循环读，直到读到产生一个EAGAIN才认为此次事件处理完成，当read()返回的读到的数据长度小于请求的数据长度时，就可以确定此时缓冲中已没有数据了，也就可以认为此事读事件已处理完成。
Level Triggered 工作模式
相反的，以LT方式调用epoll接口的时候，它就相当于一个速度比较快的poll(2)，并且无论后面的数据是否被使用，因此他们具有同样的职能。因为即使使用ET模式的epoll，在收到多个chunk的数据的时候仍然会产生多个事件。调用者可以设定EPOLLONESHOT标志，在 epoll_wait(2)收到事件后epoll会与事件关联的文件句柄从epoll描述符中禁止掉。因此当EPOLLONESHOT设定后，使用带有 EPOLL_CTL_MOD标志的epoll_ctl(2)处理文件句柄就成为调用者必须作的事情。
LT(level triggered)是epoll缺省的工作方式，并且同时支持block和no-block socket.在这种做法中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知你 的，所以，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表．
ET (edge-triggered)是高速工作方式，只支持no-block socket，它效率要比LT更高。ET与LT的区别在于，当一个新的事件到来时，ET模式下当然可以从epoll_wait调用中获取到这个事件，可是如果这次没有把这个事件对应的套接字缓冲区处理完，在这个套接字中没有新的事件再次到来时，在ET模式下是无法再次从epoll_wait调用中获取这个事件的。而LT模式正好相反，只要一个事件对应的套接字缓冲区还有数据，就总能从epoll_wait中获取这个事件。
因此，LT模式下开发基于epoll的应用要简单些，不太容易出错。而在ET模式下事件发生时，如果没有彻底地将缓冲区数据处理完，则会导致缓冲区中的用户请求得不到响应。
图示说明：
Nginx默认采用ET模式来使用epoll。
epoll的优点：
1.支持一个进程打开大数目的socket描述符(FD)
select 最不能忍受的是一个进程所打开的FD是有一定限制的，由FD_SETSIZE设置，默认值是2048。对于那些需要支持的上万连接数目的IM服务器来说显然太少了。这时候你一是可以选择修改这个宏然后重新编译内核，不过资料也同时指出这样会带来网络效率的下降，二是可以选择多进程的解决方案(传统的 Apache方案)，不过虽然linux上面创建进程的代价比较小，但仍旧是不可忽视的，加上进程间数据同步远比不上线程间同步的高效，所以也不是一种完美的方案。不过 epoll则没有这个限制，它所支持的FD上限是最大可以打开文件的数目，这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。
2.IO效率不随FD数目增加而线性下降
传统的select/poll另一个致命弱点就是当你拥有一个很大的socket集合，不过由于网络延时，任一时间只有部分的socket是"活跃"的，但是select/poll每次调用都会线性扫描全部的集合，导致效率呈现线性下降。但是epoll不存在这个问题，它只会对"活跃"的socket进行操作---这是因为在内核实现中epoll是根据每个fd上面的callback函数实现的。那么，只有"活跃"的socket才会主动的去调用 callback函数，其他idle状态socket则不会，在这点上，epoll实现了一个"伪"AIO，因为这时候推动力在os内核。在一些 benchmark中，如果所有的socket基本上都是活跃的---比如一个高速LAN环境，epoll并不比select/poll有什么效率，相反，如果过多使用epoll_ctl,效率相比还有稍微的下降。但是一旦使用idle connections模拟WAN环境,epoll的效率就远在select/poll之上了。
3.使用mmap加速内核与用户空间的消息传递
这点实际上涉及到epoll的具体实现了。无论是select,poll还是epoll都需要内核把FD消息通知给用户空间，如何避免不必要的内存拷贝就很重要，在这点上，epoll是通过内核于用户空间mmap同一块内存实现的。而如果你想我一样从2.5内核就关注epoll的话，一定不会忘记手工 mmap这一步的。
4.内核微调
这一点其实不算epoll的优点了，而是整个linux平台的优点。也许你可以怀疑linux平台，但是你无法回避linux平台赋予你微调内核的能力。比如，内核TCP/IP协议栈使用内存池管理sk_buff结构，那么可以在运行时期动态调整这个内存pool(skb_head_pool)的大小--- 通过echo XXXX>/proc/sys/net/core/hot_list_length完成。再比如listen函数的第2个参数(TCP完成3次握手的数据包队列长度)，也可以根据你平台内存大小动态调整。更甚至在一个数据包面数目巨大但同时每个数据包本身大小却很小的特殊系统上尝试最新的NAPI网卡驱动架构。
linux下epoll如何实现高效处理百万句柄的
开发高性能网络程序时，windows开发者们言必称iocp，linux开发者们则言必称epoll。大家都明白epoll是一种IO多路复用技术，可以非常高效的处理数以百万计的socket句柄，比起以前的select和poll效率高大发了。我们用起epoll来都感觉挺爽，确实快，那么，它到底为什么可以高速处理这么多并发连接呢？
使用起来很清晰，首先要调用epoll_create建立一个epoll对象。参数size是内核保证能够正确处理的最大句柄数，多于这个最大数时内核可不保证效果。
epoll_ctl可以操作上面建立的epoll，例如，将刚建立的socket加入到epoll中让其监控，或者把 epoll正在监控的某个socket句柄移出epoll，不再监控它等等。
epoll_wait在调用时，在给定的timeout时间内，当在监控的所有句柄中有事件发生时，就返回用户态的进程。
从上面的调用方式就可以看到epoll比select/poll的优越之处：因为后者每次调用时都要传递你所要监控的所有socket给select/poll系统调用，这意味着需要将用户态的socket列表copy到内核态，如果以万计的句柄会导致每次都要copy几十几百KB的内存到内核态，非常低效。而我们调用epoll_wait时就相当于以往调用select/poll，但是这时却不用传递socket句柄给内核，因为内核已经在epoll_ctl中拿到了要监控的句柄列表。
所以，实际上在你调用epoll_create后，内核就已经在内核态开始准备帮你存储要监控的句柄了，每次调用epoll_ctl只是在往内核的数据结构里塞入新的socket句柄。
当一个进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关：
/* 171 * This structure is stored inside the "private_data" member of the file 172 * structure and represents the main data structure for the eventpoll 173 * interface. 174 */ 175struct eventpoll { 176 /* Protect the access to this structure */ 177 spinlock_t lock; 178 179 /* 180 * This mutex is used to ensure that files are not removed 181 * while epoll is using them. This is held during the event 182 * collection loop, the file cleanup path, the epoll file exit 183 * code and the ctl operations. 184 */ 185 struct mutex mtx; 186 187 /* Wait queue used by sys_epoll_wait() */ 188 wait_queue_head_t wq; 189 190 /* Wait queue used by file->poll() */ 191 wait_queue_head_t poll_wait; 192 193 /* List of ready file descriptors */ 194 struct list_head rdllist; 195 196 /* RB tree root used to store monitored fd structs */ 197 struct rb_root rbr;//红黑树根节点，这棵树存储着所有添加到epoll中的事件，也就是这个epoll监控的事件 198 199 /* 200 * This is a single linked list that chains all the "struct epitem" that 201 * happened while transferring ready events to userspace w/out 202 * holding ->lock. 203 */ 204 struct epitem *ovflist; 205 206 /* wakeup_source used when ep_scan_ready_list is running */ 207 struct wakeup_source *ws; 208 209 /* The user that created the eventpoll descriptor */ 210 struct user_struct *user; 211 212 struct file *file; 213 214 /* used to optimize loop detection check */ 215 int visited; 216 struct list_head visited_list_link;//双向链表中保存着将要通过epoll_wait返回给用户的、满足条件的事件 217};
每一个epoll对象都有一个独立的eventpoll结构体，这个结构体会在内核空间中创造独立的内存，用于存储使用epoll_ctl方法向epoll对象中添加进来的事件。这样，重复的事件就可以通过红黑树而高效的识别出来。
在epoll中，对于每一个事件都会建立一个epitem结构体：
/* 130 * Each file descriptor added to the eventpoll interface will 131 * have an entry of this type linked to the "rbr" RB tree. 132 * Avoid increasing the size of this struct, there can be many thousands 133 * of these on a server and we do not want this to take another cache line. 134 */ 135struct epitem { 136 /* RB tree node used to link this structure to the eventpoll RB tree */ 137 struct rb_node rbn; 138 139 /* List header used to link this structure to the eventpoll ready list */ 140 struct list_head rdllink; 141 142 /* 143 * Works together "struct eventpoll"->ovflist in keeping the 144 * single linked chain of items. 145 */ 146 struct epitem *next; 147 148 /* The file descriptor information this item refers to */ 149 struct epoll_filefd ffd; 150 151 /* Number of active wait queue attached to poll operations */ 152 int nwait; 153 154 /* List containing poll wait queues */ 155 struct list_head pwqlist; 156 157 /* The "container" of this item */ 158 struct eventpoll *ep; 159 160 /* List header used to link this item to the "struct file" items list */ 161 struct list_head fllink; 162 163 /* wakeup_source used when EPOLLWAKEUP is set */ 164 struct wakeup_source __rcu *ws; 165 166 /* The structure that describe the interested events and the source fd */ 167 struct epoll_event event; 168};
此外，epoll还维护了一个双链表，用户存储发生的事件。当epoll_wait调用时，仅仅观察这个list链表里有没有数据即eptime项即可。有数据就返回，没有数据就sleep，等到timeout时间到后即使链表没数据也返回。所以，epoll_wait非常高效。
而且，通常情况下即使我们要监控百万计的句柄，大多一次也只返回很少量的准备就绪句柄而已，所以，epoll_wait仅需要从内核态copy少量的句柄到用户态而已，如何能不高效？！
那么，这个准备就绪list链表是怎么维护的呢？当我们执行epoll_ctl时，除了把socket放到epoll文件系统里file对象对应的红黑树上之外，还会给内核中断处理程序注册一个回调函数，告诉内核，如果这个句柄的中断到了，就把它放到准备就绪list链表里。所以，当一个socket上有数据到了，内核在把网卡上的数据copy到内核中后就来把socket插入到准备就绪链表里了。
如此，一颗红黑树，一张准备就绪句柄链表，少量的内核cache，就帮我们解决了大并发下的socket处理问题。执行epoll_create时，创建了红黑树和就绪链表，执行epoll_ctl时，如果增加socket句柄，则检查在红黑树中是否存在，存在立即返回，不存在则添加到树干上，然后向内核注册回调函数，用于当中断事件来临时向准备就绪链表中插入数据。执行epoll_wait时立刻返回准备就绪链表里的数据即可。
epoll的使用方法
那么究竟如何来使用epoll呢？其实非常简单。
通过在包含一个头文件#include <sys/epoll.h> 以及几个简单的API将可以大大的提高你的网络服务器的支持人数。
首先通过create_epoll(int maxfds)来创建一个epoll的句柄。这个函数会返回一个新的epoll句柄，之后的所有操作将通过这个句柄来进行操作。在用完之后，记得用close()来关闭这个创建出来的epoll句柄。
之后在你的网络主循环里面，每一帧的调用epoll_wait(int epfd, epoll_event events, int max events, int timeout)来查询所有的网络接口，看哪一个可以读，哪一个可以写了。基本的语法为：
nfds = epoll_wait(kdpfd, events, maxevents, -1);
其中kdpfd为用epoll_create创建之后的句柄，events是一个epoll_event*的指针，当epoll_wait这个函数操作成功之后，epoll_events里面将储存所有的读写事件。max_events是当前需要监听的所有socket句柄数。最后一个timeout是 epoll_wait的超时，为0的时候表示马上返回，为-1的时候表示一直等下去，直到有事件返回，为任意正整数的时候表示等这么长的时间，如果一直没有事件，则返回。一般如果网络主循环是单独的线程的话，可以用-1来等，这样可以保证一些效率，如果是和主逻辑在同一个线程的话，则可以用0来保证主循环的效率。
epoll_wait返回之后应该是一个循环，遍历所有的事件。
几乎所有的epoll程序都使用下面的框架：
for( ; ; ) { nfds = epoll_wait(epfd,events,20,500); for(i=0;i fd; send( sockfd, md->ptr, strlen((char*)md->ptr), 0 ); //发送数据 ev.data.fd=sockfd; ev.events=EPOLLIN|EPOLLET; epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); //修改标识符，等待下一个循环时接收数据 } else { //其他的处理 } } }
epoll的程序实例1
#include#include #include #include #include #include #include #include #include #define MAXEVENTS 64 //函数: //功能:创建和绑定一个TCP socket //参数:端口 //返回值:创建的socket static int create_and_bind (char *port) { struct addrinfo hints; struct addrinfo *result, *rp; int s, sfd; memset (&hints, 0, sizeof (struct addrinfo)); hints.ai_family = AF_UNSPEC; /* Return IPv4 and IPv6 choices */ hints.ai_socktype = SOCK_STREAM; /* We want a TCP socket */ hints.ai_flags = AI_PASSIVE; /* All interfaces */ s = getaddrinfo (NULL, port, &hints, &result); if (s != 0) { fprintf (stderr, "getaddrinfo: %s\n", gai_strerror (s)); return -1; } for (rp = result; rp != NULL; rp = rp->ai_next) { sfd = socket (rp->ai_family, rp->ai_socktype, rp->ai_protocol); if (sfd == -1) continue; s = bind (sfd, rp->ai_addr, rp->ai_addrlen); if (s == 0) { /* We managed to bind successfully! */ break; } close (sfd); } if (rp == NULL) { fprintf (stderr, "Could not bind\n"); return -1; } freeaddrinfo (result); return sfd; } //函数 //功能:设置socket为非阻塞的 static int make_socket_non_blocking (int sfd) { int flags, s; //得到文件状态标志 flags = fcntl (sfd, F_GETFL, 0); if (flags == -1) { perror ("fcntl"); return -1; } //设置文件状态标志 flags |= O_NONBLOCK; s = fcntl (sfd, F_SETFL, flags); if (s == -1) { perror ("fcntl"); return -1; } return 0; } //端口由参数argv[1]指定 int main (int argc, char *argv[]) { int sfd, s; int efd; struct epoll_event event; struct epoll_event *events; if (argc != 2) { fprintf (stderr, "Usage: %s [port]\n", argv[0]); exit (EXIT_FAILURE); } sfd = create_and_bind (argv[1]); if (sfd == -1) abort (); s = make_socket_non_blocking (sfd); if (s == -1) abort (); s = listen (sfd, SOMAXCONN); if (s == -1) { perror ("listen"); abort (); } //除了参数size被忽略外,此函数和epoll_create完全相同 efd = epoll_create1 (0); if (efd == -1) { perror ("epoll_create"); abort (); } event.data.fd = sfd; event.events = EPOLLIN | EPOLLET;//读入,边缘触发方式 s = epoll_ctl (efd, EPOLL_CTL_ADD, sfd, &event); if (s == -1) { perror ("epoll_ctl"); abort (); } /* Buffer where events are returned */ events = calloc (MAXEVENTS, sizeof event); /* The event loop */ while (1) { int n, i; n = epoll_wait (efd, events, MAXEVENTS, -1); for (i = 0; i < n; i++) { if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP) || (!(events[i].events & EPOLLIN))) { /* An error has occured on this fd, or the socket is not ready for reading (why were we notified then?) */ fprintf (stderr, "epoll error\n"); close (events[i].data.fd); continue; } else if (sfd == events[i].data.fd) { /* We have a notification on the listening socket, which means one or more incoming connections. */ while (1) { struct sockaddr in_addr; socklen_t in_len; int infd; char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV]; in_len = sizeof in_addr; infd = accept (sfd, &in_addr, &in_len); if (infd == -1) { if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) { /* We have processed all incoming connections. */ break; } else { perror ("accept"); break; } } //将地址转化为主机名或者服务名 s = getnameinfo (&in_addr, in_len, hbuf, sizeof hbuf, sbuf, sizeof sbuf, NI_NUMERICHOST | NI_NUMERICSERV);//flag参数:以数字名返回 //主机地址和服务地址 if (s == 0) { printf("Accepted connection on descriptor %d " "(host=%s, port=%s)\n", infd, hbuf, sbuf); } /* Make the incoming socket non-blocking and add it to the list of fds to monitor. */ s = make_socket_non_blocking (infd); if (s == -1) abort (); event.data.fd = infd; event.events = EPOLLIN | EPOLLET; s = epoll_ctl (efd, EPOLL_CTL_ADD, infd, &event); if (s == -1) { perror ("epoll_ctl"); abort (); } } continue; } else { /* We have data on the fd waiting to be read. Read and display it. We must read whatever data is available completely, as we are running in edge-triggered mode and won't get a notification again for the same data. */ int done = 0; while (1) { ssize_t count; char buf[512]; count = read (events[i].data.fd, buf, sizeof(buf)); if (count == -1) { /* If errno == EAGAIN, that means we have read all data. So go back to the main loop. */ if (errno != EAGAIN) { perror ("read"); done = 1; } break; } else if (count == 0) { /* End of file. The remote has closed the connection. */ done = 1; break; } /* Write the buffer to standard output */ s = write (1, buf, count); if (s == -1) { perror ("write"); abort (); } } if (done) { printf ("Closed connection on descriptor %d\n", events[i].data.fd); /* Closing the descriptor will make epoll remove it from the set of descriptors which are monitored. */ close (events[i].data.fd); } } } } free (events); close (sfd); return EXIT_SUCCESS; }
运行方式：
在一个终端运行此程序：epoll.out PORT
另一个终端：telnet 127.0.0.1 PORT
截图：
epoll的程序实例2
支持多客户端与一个服务端进行通信，客户端给服务端发送字符串数据，服务端将字符串中小写转为大写后发送回客户端，客户端打印输出经转换后的字符串。
例如：发送
abcde，打印输出ABCDE 服务端源码如下：
/*server.c*/#include#include #include #include #include #include #include #define BUFSIZE 666#define SERV_PORT 8000#define OPEN_MAX 1024int main(){ int i, j, maxi, listenfd, connfd, sockfd; int nready, efd, res; ssize_t n; char buf[BUFSIZE], str[INET_ADDRSTRLEN]; socklen_t clilen; int client[OPEN_MAX]; struct sockaddr_in cliaddr, servaddr; struct epoll_event tep, ep[OPEN_MAX];//监听事件 /*分配一个网络通信套接字，监听文件描述符listenfd*/ listenfd = socket(AF_INET, SOCK_STREAM, 0); /*初始化 IP类型 端口*/ bzero(&servaddr, sizeof(servaddr)); servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); servaddr.sin_port = htons(SERV_PORT); /*将listenfd绑定服务端地址*/ bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)); /*监听请求*/ listen(listenfd, 20); /*将客户端标识初始化为-1*/ for(i = 0; i < OPEN_MAX; i++){ client[i] = -1; } maxi = -1; /*告诉内核要监听的文件描述符个数 OPEN_MAX = 1024*/ efd = epoll_create(OPEN_MAX); if(efd == -1){ perror("epoll_create"); } tep.events = EPOLLIN;/*监听文件描述符的可读事件*/ tep.data.fd = listenfd;/*设置为监听的文件描述符*/ /*控制epoll监控的文件描述符上的事件*/ res = epoll_ctl(efd, EPOLL_CTL_ADD/*注册新的fd到efd*/, listenfd, &tep); if(res == -1) perror("epoll_ctl"); for(;;){ /*等待所监控文件描述符上有事件的产生，阻塞监听*/ nready = epoll_wait(efd, ep, OPEN_MAX, -1); if(nready == -1) perror("epoll_wait"); for(i = 0; i < nready; i++){ if(!(ep[i].events & EPOLLIN))/*若不是EPOLLIN事件，不做往下的处理*/ continue; if(ep[i].data.fd == listenfd){ /*若是EPOLLIN，执行连接，接受请求*/ clilen = sizeof(cliaddr); /*接受请求，分配新文件描述符connfd进行通信*/ connfd = accept(listenfd, (struct sockaddr*)&cliaddr, &clilen); printf("received from %s at PORT %d\n", (char*)inet_ntop(AF_INET, &cliaddr.sin_addr, str, sizeof(str)), ntohs(cliaddr.sin_port)); /*若将此新客户端添加至客户端集中*/ for(j = 0; j < OPEN_MAX; j++) if(client[j] < 0){ client[j] = connfd; break; } if(j == OPEN_MAX) perror("客户端超过限制"); if(j > maxi) maxi = j;//保证maxi为最大文件描述符 tep.events = EPOLLIN; tep.data.fd = connfd; res = epoll_ctl(efd, EPOLL_CTL_ADD/*注册新的connfd到efd*/, connfd, &tep); if(res == -1) perror("epoll_ctl"); }else{/*处理efd中监听的客户端请求*/ sockfd = ep[i].data.fd; n = read(sockfd, buf, BUFSIZE); if(n == 0){ /*读取若为空*/ for(j = 0; j <= maxi; j++){ if(client[j] == sockfd){ client[j] = -1; break; } } /*清除对sockfd文件描述符事件的监听*/ res = epoll_ctl(efd, EPOLL_CTL_DEL, sockfd, NULL); if(res == -1) perror("epoll_ctl"); close(sockfd); printf("client[%d] closed connection\n", j); }else{/*非空则处理客户端信息*/ for(j = 0; j
客户端源码如下：
/*client.c*/#include#include #include #include #include #include #include #include #define BUFSIZE 666#define SERV_PORT 8000#define OPEN_MAX 1024int main(int argc, char *argv[]){ struct sockaddr_in servaddr; char buf[BUFSIZE]; int sockfd, n; sockfd = socket(AF_INET, SOCK_STREAM, 0); bzero(&servaddr, sizeof(servaddr)); servaddr.sin_family = AF_INET; inet_pton(AF_INET, "127.0.0.1", &servaddr.sin_addr); servaddr.sin_port = htons(SERV_PORT); /*连接服务端*/ connect(sockfd, (struct sockaddr*)&servaddr, sizeof(servaddr)); while(fgets(buf, BUFSIZE, stdin) != NULL){ /*通过sockfd给服务端发送数据*/ write(sockfd, buf, strlen(buf)); n = read(sockfd, buf, BUFSIZE); if(n == 0) printf("the other side has been closed.\n"); else/*打印输出服务端传过来的数据*/ write(STDOUT_FILENO, buf, n); } close(sockfd); return 0;}
编译及执行
在终端1先执行：
另开一终端2：
再另开一终端3：
最后开一终端4：
关闭3个客户端(CTR+C)后服务端窗口：
参考资料：
《深入理解Nginx模块开发与架构解析》9.6小节
IOCP模型与网络编程
1) 使用IOCP模型编程的优点
① 帮助维持重复使用的内存池。(与重叠I/O技术有关)
② 去除删除线程创建/终结负担。
③ 利于管理，分配线程，控制并发，最小化的线程上下文切换。
④ 优化线程调度，提高CPU和内存缓冲的命中率。
2) 使用IOCP模型编程汲及到的知识点（无先后顺序）
① 同步与异步
② 阻塞与非阻塞
③ 重叠I/O技术
④ 多线程
⑤ 栈、队列这两种基本的数据结构
3) 需要使用上的API函数
① 与SOCKET相关
1、链接套接字动态链接库：int WSAStartup(...);
2、创建套接字库： SOCKET socket(...);
3、绑字套接字： int bind(...);
4、套接字设为监听状态： int listen(...);
5、接收套接字： SOCKET accept(...);
6、向指定套接字发送信息：int send(...);
7、从指定套接字接收信息：int recv(...);
② 与线程相关
1、创建线程：HANDLE CreateThread(...);
③ 重叠I/O技术相关
1、向套接字发送数据： int WSASend(...);
2、向套接字发送数据包： int WSASendFrom(...);
3、从套接字接收数据： int WSARecv(...);
4、从套接字接收数据包： int WSARecvFrom(...);
④ IOCP相关
1、创建完成端口： HANDLE WINAPI CreateIoCompletionPort(...);
2、关联完成端口： HANDLE WINAPI CreateIoCompletionPort(...);
3、获取队列完成状态: BOOL WINAPI GetQueuedCompletionStatus(...);
4、投递一个队列完成状态：BOOL WINAPI PostQueuedCompletionStatus(...);
完整的简单的IOCP服务器与客户端代码实例：
// IOCP_TCPIP_Socket_Server.cpp#include#include #include #include using namespace std;#pragma comment(lib, "Ws2_32.lib") // Socket编程需用的动态链接库#pragma comment(lib, "Kernel32.lib") // IOCP需要用到的动态链接库/** * 结构体名称：PER_IO_DATA * 结构体功能：重叠I/O需要用到的结构体，临时记录IO数据 **/const int DataBuffSize = 2 * 1024;typedef struct{ OVERLAPPED overlapped; WSABUF databuff; char buffer[ DataBuffSize ]; int BufferLen; int operationType;}PER_IO_OPERATEION_DATA, *LPPER_IO_OPERATION_DATA, *LPPER_IO_DATA, PER_IO_DATA;/** * 结构体名称：PER_HANDLE_DATA * 结构体存储：记录单个套接字的数据，包括了套接字的变量及套接字的对应的客户端的地址。 * 结构体作用：当服务器连接上客户端时，信息存储到该结构体中，知道客户端的地址以便于回访。 **/typedef struct{ SOCKET socket; SOCKADDR_STORAGE ClientAddr;}PER_HANDLE_DATA, *LPPER_HANDLE_DATA;// 定义全局变量const int DefaultPort = 6000; vector < PER_HANDLE_DATA* > clientGroup; // 记录客户端的向量组HANDLE hMutex = CreateMutex(NULL, FALSE, NULL);DWORD WINAPI ServerWorkThread(LPVOID CompletionPortID);DWORD WINAPI ServerSendThread(LPVOID IpParam);// 开始主函数int main(){// 加载socket动态链接库 WORD wVersionRequested = MAKEWORD(2, 2); // 请求2.2版本的WinSock库 WSADATA wsaData; // 接收Windows Socket的结构信息 DWORD err = WSAStartup(wVersionRequested, &wsaData); if (0 != err){ // 检查套接字库是否申请成功 cerr << "Request Windows Socket Library Error!\n"; system("pause"); return -1; } if(LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2){// 检查是否申请了所需版本的套接字库 WSACleanup(); cerr << "Request Windows Socket Version 2.2 Error!\n"; system("pause"); return -1; }// 创建IOCP的内核对象 /** * 需要用到的函数的原型： * HANDLE WINAPI CreateIoCompletionPort( * __in HANDLE FileHandle, // 已经打开的文件句柄或者空句柄，一般是客户端的句柄 * __in HANDLE ExistingCompletionPort, // 已经存在的IOCP句柄 * __in ULONG_PTR CompletionKey, // 完成键，包含了指定I/O完成包的指定文件 * __in DWORD NumberOfConcurrentThreads // 真正并发同时执行最大线程数，一般推介是CPU核心数*2 * ); **/ HANDLE completionPort = CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0); if (NULL == completionPort){ // 创建IO内核对象失败 cerr << "CreateIoCompletionPort failed. Error:" << GetLastError() << endl; system("pause"); return -1; }// 创建IOCP线程--线程里面创建线程池 // 确定处理器的核心数量 SYSTEM_INFO mySysInfo; GetSystemInfo(&mySysInfo); // 基于处理器的核心数量创建线程 for(DWORD i = 0; i < (mySysInfo.dwNumberOfProcessors * 2); ++i){ // 创建服务器工作器线程，并将完成端口传递到该线程 HANDLE ThreadHandle = CreateThread(NULL, 0, ServerWorkThread, completionPort, 0, NULL); if(NULL == ThreadHandle){ cerr << "Create Thread Handle failed. Error:" << GetLastError() << endl; system("pause"); return -1; } CloseHandle(ThreadHandle); }// 建立流式套接字 SOCKET srvSocket = socket(AF_INET, SOCK_STREAM, 0);// 绑定SOCKET到本机 SOCKADDR_IN srvAddr; srvAddr.sin_addr.S_un.S_addr = htonl(INADDR_ANY); srvAddr.sin_family = AF_INET; srvAddr.sin_port = htons(DefaultPort); int bindResult = bind(srvSocket, (SOCKADDR*)&srvAddr, sizeof(SOCKADDR)); if(SOCKET_ERROR == bindResult){ cerr << "Bind failed. Error:" << GetLastError() << endl; system("pause"); return -1; }// 将SOCKET设置为监听模式 int listenResult = listen(srvSocket, 10); if(SOCKET_ERROR == listenResult){ cerr << "Listen failed. Error: " << GetLastError() << endl; system("pause"); return -1; } // 开始处理IO数据 cout << "本服务器已准备就绪，正在等待客户端的接入...\n"; // 创建用于发送数据的线程 HANDLE sendThread = CreateThread(NULL, 0, ServerSendThread, 0, 0, NULL); while(true){ PER_HANDLE_DATA * PerHandleData = NULL; SOCKADDR_IN saRemote; int RemoteLen; SOCKET acceptSocket; // 接收连接，并分配完成端，这儿可以用AcceptEx() RemoteLen = sizeof(saRemote); acceptSocket = accept(srvSocket, (SOCKADDR*)&saRemote, &RemoteLen); if(SOCKET_ERROR == acceptSocket){ // 接收客户端失败 cerr << "Accept Socket Error: " << GetLastError() << endl; system("pause"); return -1; } // 创建用来和套接字关联的单句柄数据信息结构 PerHandleData = (LPPER_HANDLE_DATA)GlobalAlloc(GPTR, sizeof(PER_HANDLE_DATA)); // 在堆中为这个PerHandleData申请指定大小的内存 PerHandleData -> socket = acceptSocket; memcpy (&PerHandleData -> ClientAddr, &saRemote, RemoteLen); clientGroup.push_back(PerHandleData); // 将单个客户端数据指针放到客户端组中 // 将接受套接字和完成端口关联 CreateIoCompletionPort((HANDLE)(PerHandleData -> socket), completionPort, (DWORD)PerHandleData, 0); // 开始在接受套接字上处理I/O使用重叠I/O机制 // 在新建的套接字上投递一个或多个异步 // WSARecv或WSASend请求，这些I/O请求完成后，工作者线程会为I/O请求提供服务 // 单I/O操作数据(I/O重叠) LPPER_IO_OPERATION_DATA PerIoData = NULL; PerIoData = (LPPER_IO_OPERATION_DATA)GlobalAlloc(GPTR, sizeof(PER_IO_OPERATEION_DATA)); ZeroMemory(&(PerIoData -> overlapped), sizeof(OVERLAPPED)); PerIoData->databuff.len = 1024; PerIoData->databuff.buf = PerIoData->buffer; PerIoData->operationType = 0; // read DWORD RecvBytes; DWORD Flags = 0; WSARecv(PerHandleData->socket, &(PerIoData->databuff), 1, &RecvBytes, &Flags, &(PerIoData->overlapped), NULL); } system("pause"); return 0;}// 开始服务工作线程函数DWORD WINAPI ServerWorkThread(LPVOID IpParam){ HANDLE CompletionPort = (HANDLE)IpParam; DWORD BytesTransferred; LPOVERLAPPED IpOverlapped; LPPER_HANDLE_DATA PerHandleData = NULL; LPPER_IO_DATA PerIoData = NULL; DWORD RecvBytes; DWORD Flags = 0; BOOL bRet = false; while(true){ bRet = GetQueuedCompletionStatus(CompletionPort, &BytesTransferred, (PULONG_PTR)&PerHandleData, (LPOVERLAPPED*)&IpOverlapped, INFINITE); if(bRet == 0){ cerr << "GetQueuedCompletionStatus Error: " << GetLastError() << endl; return -1; } PerIoData = (LPPER_IO_DATA)CONTAINING_RECORD(IpOverlapped, PER_IO_DATA, overlapped); // 检查在套接字上是否有错误发生 if(0 == BytesTransferred){ closesocket(PerHandleData->socket); GlobalFree(PerHandleData); GlobalFree(PerIoData); continue; } // 开始数据处理，接收来自客户端的数据 WaitForSingleObject(hMutex,INFINITE); cout << "A Client says: " << PerIoData->databuff.buf << endl; ReleaseMutex(hMutex); // 为下一个重叠调用建立单I/O操作数据 ZeroMemory(&(PerIoData->overlapped), sizeof(OVERLAPPED)); // 清空内存 PerIoData->databuff.len = 1024; PerIoData->databuff.buf = PerIoData->buffer; PerIoData->operationType = 0; // read WSARecv(PerHandleData->socket, &(PerIoData->databuff), 1, &RecvBytes, &Flags, &(PerIoData->overlapped), NULL); } return 0;}// 发送信息的线程执行函数DWORD WINAPI ServerSendThread(LPVOID IpParam){ while(1){ char talk[200]; gets(talk); int len; for (len = 0; talk[len] != '\0'; ++len){ // 找出这个字符组的长度 } talk[len] = '\n'; talk[++len] = '\0'; printf("I Say:"); cout << talk; WaitForSingleObject(hMutex,INFINITE); for(int i = 0; i < clientGroup.size(); ++i){ send(clientGroup[i]->socket, talk, 200, 0); // 发送信息 } ReleaseMutex(hMutex); } return 0;}
// IOCP_TCPIP_Socket_Client.cpp#include#include #include #include #include #include using namespace std;#pragma comment(lib, "Ws2_32.lib") // Socket编程需用的动态链接库SOCKET sockClient; // 连接成功后的套接字HANDLE bufferMutex; // 令其能互斥成功正常通信的信号量句柄const int DefaultPort = 6000;int main(){// 加载socket动态链接库(dll) WORD wVersionRequested; WSADATA wsaData; // 这结构是用于接收Wjndows Socket的结构信息的 wVersionRequested = MAKEWORD( 2, 2 ); // 请求2.2版本的WinSock库 int err = WSAStartup( wVersionRequested, &wsaData ); if ( err != 0 ) { // 返回值为零的时候是表示成功申请WSAStartup return -1; } if ( LOBYTE( wsaData.wVersion ) != 2 || HIBYTE( wsaData.wVersion ) != 2 ) { // 检查版本号是否正确 WSACleanup( ); return -1; } // 创建socket操作，建立流式套接字，返回套接字号sockClient sockClient = socket(AF_INET, SOCK_STREAM, 0); if(sockClient == INVALID_SOCKET) { printf("Error at socket():%ld\n", WSAGetLastError()); WSACleanup(); return -1; } // 将套接字sockClient与远程主机相连 // int connect( SOCKET s, const struct sockaddr* name, int namelen); // 第一个参数：需要进行连接操作的套接字 // 第二个参数：设定所需要连接的地址信息 // 第三个参数：地址的长度 SOCKADDR_IN addrSrv; addrSrv.sin_addr.S_un.S_addr = inet_addr("127.0.0.1"); // 本地回路地址是127.0.0.1; addrSrv.sin_family = AF_INET; addrSrv.sin_port = htons(DefaultPort); while(SOCKET_ERROR == connect(sockClient, (SOCKADDR*)&addrSrv, sizeof(SOCKADDR))){ // 如果还没连接上服务器则要求重连 cout << "服务器连接失败，是否重新连接？（Y/N):"; char choice; while(cin >> choice && (!((choice != 'Y' && choice == 'N') || (choice == 'Y' && choice != 'N')))){ cout << "输入错误，请重新输入:"; cin.sync(); cin.clear(); } if (choice == 'Y'){ continue; } else{ cout << "退出系统中..."; system("pause"); return 0; } } cin.sync(); cout << "本客户端已准备就绪，用户可直接输入文字向服务器反馈信息。\n"; send(sockClient, "\nAttention: A Client has enter...\n", 200, 0); bufferMutex = CreateSemaphore(NULL, 1, 1, NULL); DWORD WINAPI SendMessageThread(LPVOID IpParameter); DWORD WINAPI ReceiveMessageThread(LPVOID IpParameter); HANDLE sendThread = CreateThread(NULL, 0, SendMessageThread, NULL, 0, NULL); HANDLE receiveThread = CreateThread(NULL, 0, ReceiveMessageThread, NULL, 0, NULL); WaitForSingleObject(sendThread, INFINITE); // 等待线程结束 closesocket(sockClient); CloseHandle(sendThread); CloseHandle(receiveThread); CloseHandle(bufferMutex); WSACleanup(); // 终止对套接字库的使用 printf("End linking...\n"); printf("\n"); system("pause"); return 0;}DWORD WINAPI SendMessageThread(LPVOID IpParameter){ while(1){ string talk; getline(cin, talk); WaitForSingleObject(bufferMutex, INFINITE); // P（资源未被占用） if("quit" == talk){ talk.push_back('\0'); send(sockClient, talk.c_str(), 200, 0); break; } else{ talk.append("\n"); } printf("\nI Say:(\"quit\"to exit):"); cout << talk; send(sockClient, talk.c_str(), 200, 0); // 发送信息 ReleaseSemaphore(bufferMutex, 1, NULL); // V（资源占用完毕） } return 0;}DWORD WINAPI ReceiveMessageThread(LPVOID IpParameter){ while(1){ char recvBuf[300]; recv(sockClient, recvBuf, 200, 0); WaitForSingleObject(bufferMutex, INFINITE); // P（资源未被占用） printf("%s Says: %s", "Server", recvBuf); // 接收信息 ReleaseSemaphore(bufferMutex, 1, NULL); // V（资源占用完毕） } return 0;}
一：IOCP和Epoll之间的异同。
异： 1：IOCP是WINDOWS系统下使用。Epoll是Linux系统下使用。 2：IOCP是IO操作完毕之后，通过Get函数获得一个完成的事件通知。 Epoll是当你希望进行一个IO操作时，向Epoll查询是否可读或者可写，若处于可读或可写状态后，Epoll会通过epoll_wait进行通知。 3：IOCP封装了异步的消息事件的通知机制，同时封装了部分IO操作。但Epoll仅仅封装了一个异步事件的通知机制，并不负责IO读写操作。Epoll保持了事件通知和IO操作间的独立性，更加简单灵活。 4： 基于上面的描述，我们可以知道Epoll不负责IO操作，所以它只告诉你当前可读可写了，并且将协议读写缓冲填充，由用户去读写控制，此时我们可以做出额 外的许多操作。IOCP则直接将IO通道里的读写操作都做完了才通知用户，当IO通道里发生了堵塞等状况我们是无法控制的。同：
1：它们都是异步的事件驱动的网络模型。 2：它们都可以向底层进行指针数据传递，当返回事件时，除可通知事件类型外，还可以通知事件相关数据。二：描述一下IOCP:
扯远点。首先传统服务器的网络IO流程如下： 接到一个客户端连接->创建一个线程负责这个连接的IO操作->持续对新线程进行数据处理->全部数据处理完毕->终止线程。 但是这样的设计代价是： 1：每个连接创建一个线程，将导致过多的线程。 2：维护线程所消耗的堆栈内存过大。 3：操作系统创建和销毁线程过大。 4：线程之间切换的上下文代价过大。 此时我们可以考虑使用线程池解决其中3和4的问题。这种传统的服务器网络结构称之为会话模型。 后来我们为防止大量线程的维护，创建了I/O模型，它被希望要求可以: 1：允许一个线程在不同时刻给多个客户端进行服务。 2：允许一个客户端在不同时间被多个线程服务。 这样做的话，我们的线程则会大幅度减少，这就要求以下两点： 1：客户端状态的分离，之前会话模式我们可以通过线程状态得知客户端状态，但现在客户端状态要通过其他方式获取。 2：I/O请求的分离。一个线程不再服务于一个客户端会话，则要求客户端对这个线程提交I/O处理请求。 那么就产生了这样一个模式，分为两部分: 1：会话状态管理模块。它负责接收到一个客户端连接，就创建一个会话状态。 2：当会话状态发生改变，例如断掉连接，接收到网络消息，就发送一个I/O请求给 I/O工作模块进行处理。 3：I/O工作模块接收到一个I/O请求后，从线程池里唤醒一个工作线程，让该工作线程处理这个I/O请求，处理完毕后，该工作线程继续挂起。 上面的做法，则将网络连接 和I/O工作线程分离为两个部分，相互通讯仅依靠 I/O请求。 此时可知有以下一些建议： 1：在进行I/O请求处理的工作线程是被唤醒的工作线程，一个CPU对应一个的话，可以最大化利用CPU。所以 活跃线程的个数 建议等于 硬件CPU个数。 2：工作线程我们开始创建了线程池，免除创建和销毁线程的代价。因为线程是对I/O进行操作的，且一一对应，那么当I/O全部并行时，工作线程必须满足I/O并行操作需求，所以 线程池内最大工作线程个数 建议大于或者等于 I/O并行个数。 3：但是我们可知CPU个数又限制了活跃的线程个数，那么线程池过大意义很低，所以按常规建议 线程池大小 等于 CPU个数*2 左右为佳。例如，8核服务器建议创建16个工作线程的线程池。 上面描述的依然是I/O模型并非IOCP，那么IOCP是什么呢，全称 IO完成端口。 它是一种WIN32的网络I/O模型，既包括了网络连接部分，也负责了部分的I/O操作功能，用于方便我们控制有并发性的网络I/O操作。它有如下特点： 1：它是一个WIN32内核对象，所以无法运行于Linux. 2：它自己负责维护了工作线程池，同时也负责了I/O通道的内存池。 3：它自己实现了线程的管理以及I/O请求通知，最小化的做到了线程的上下文切换。 4：它自己实现了线程的优化调度，提高了CPU和内存缓冲的使用率。 使用IOCP的基本步骤很简单： 1：创建IOCP对象，由它负责管理多个Socket和I/O请求。CreateIoCompletionPort需要将IOCP对象和IOCP句柄绑定。 2：创建一个工作线程池，以便Socket发送I/O请求给IOCP对象后，由这些工作线程进行I/O操作。注意，创建这些线程的时候，将这些线程绑定到IOCP上。 3：创建一个监听的socket。 4：轮询，当接收到了新的连接后，将socket和完成端口进行关联并且投递给IOCP一个I/O请求。注意：将Socket和IOCP进行关联的函数和创建IOCP的函数一样，都是CreateIoCompletionPort，不过注意传参必然是不同的。 5：因为是异步的，我们可以去做其他，等待IOCP将I/O操作完成会回馈我们一个消息，我们再进行处理。 其中需要知道的是：I/O请求被放在一个I/O请求队列里面，对，是队列，LIFO机制。当一个设备处理完I/O请求后，将会将这个完成后的I/O请求丢回IOCP的I/O完成队列。 我们应用程序则需要在GetQueuedCompletionStatus去询问IOCP，该I/O请求是否完成。 其中有一些特殊的事情要说明一下，我们有时有需要人工的去投递一些I/O请求，则需要使用PostQueuedCompletionStatus函数向IOCP投递一个I/O请求到它的请求队列中。三：网络游戏服务器注意事项，优化措施 1：IO操作是最大的性能消耗点，注意优化余地很大。 2：算法数据结构。排序寻路算法的优化。list,vector,hashmap的选择。大数据寻址，不要考虑遍历，注意考虑hash. 3：内存管理。重载new/delete，内存池，对象池的处理。 4：数据的提前准备和即时计算。 5：CPU方面的统计监视。逻辑帧计数（应当50ms以内）。 6：预分配池减少切换和调度，预处理的线程池和连接池等。 7：基与消息队列的统计和信息监视框架。 8：CPU消耗排名：第一AOI同步，第二网络发包I/O操作，第三技能/BUFF判定计算处理，第四定时器的频率。 9：内存泄露检测，内存访问越界警惕，内存碎片的回收。 10：内存消耗排名：第一玩家对象包括其物品，第二网络数据缓冲。 11：注意32位和64位的内存容错。 12：减少不必要的分包发送。 13：减少重复包和重拷贝包的代价。 14：建议分紧急包（立刻发送）和非紧急包（定时轮训发送）。 15：带宽消耗排名：第一移动位置同步，第二对象加载，第三登陆突发包，第四状态机定时器消息。 16：客户端可做部分预判断机制，部分操作尽量分包发送。 17：大量玩家聚集时，部分非紧急包进行丢弃。 18：注意数据库单表内key数量。 19：活跃用户和非活跃用户的分割存取处理。 20：控制玩家操作对数据库的操作频率。 21：注意使用共享内存等方式对数据进行安全备份存储。 22：注意安全策略，对内网进行IP检查，对日志进行记录，任意两环点内均使用加密算法会更佳。 23：实时注意对网关，数据库等接口进行监察控制。 24：定时器应当存储一个队列，而非单向定位。 25：九宫格数据同步时，不需要直接进行九宫格的同步，对角色加一个AOI，基于圆方碰撞原理，抛弃不必要的格信息，可大幅节省。 26：客户端做部分的预测机制，服务器检测时注意时间戳问题。 27：定期心跳包，检查死链接是必要的。 28：为了实现更加负责多种类的AI，AI寻路独立服务器设计已经是必须的了。其次需要考虑的是聊天，同步。 29：服务器内网间可以考虑使用UDP。 30：注意所有内存池，对象池等的动态扩张分配。
1：以内存换取CPU的理念。
2：NPC不死理念。(只会disable) 3：动态扩展理念，负载均衡理念。 4：客户端不可信理念。 5：指针数据，消息均不可信理念。
介绍几种常见的I/O模型及其区别，如下：
-
blocking I/O
-
nonblocking I/O
-
I/O multiplexing (select and poll)
-
signal driven I/O (SIGIO)
-
asynchronous I/O (the POSIX aio_functions)
blocking I/O
这个不用多解释吧，阻塞套接字。下图是它调用过程的图示：
重点解释下上图，下面例子都会讲到。首先application调用 recvfrom()转入kernel，注意kernel有2个过程，wait for data和copy data from kernel to user。直到最后copy complete后，recvfrom()才返回。此过程一直是阻塞的。
nonblocking I/O：
与blocking I/O对立的，非阻塞套接字，调用过程图如下：
可以看见，如果直接操作它，那就是个轮询。。直到内核缓冲区有数据。
I/O multiplexing (select and poll)
最常见的I/O复用模型，select。
select先阻塞，有活动套接字才返回。与blocking I/O相比，select会有两次系统调用，但是select能处理多个套接字。
signal driven I/O (SIGIO)
只有UNIX系统支持，感兴趣的课查阅相关资料
与I/O multiplexing (select and poll)相比，它的优势是，免去了select的阻塞与轮询，当有活跃套接字时，由注册的handler处理。
asynchronous I/O (the POSIX aio_functions)
很少有*nix系统支持，windows的IOCP则是此模型
完全异步的I/O复用机制，因为纵观上面其它四种模型，至少都会在由kernel copy data to appliction时阻塞。而该模型是当copy完成后才通知application，可见是纯异步的。好像只有windows的完成端口是这个模型，效率也很出色。
下面是以上五种模型的比较
可以看出，越往后，阻塞越少，理论上效率也是最优。
=====================分割线==================================
5种模型的比较比较清晰了，剩下的就是把select,epoll,iocp,kqueue按号入座那就OK了。
poll 模型：I/O多路复用技术。poll模型将不会受限于FD_SETSIZE，因为内核所扫描的文件 描述符集合的大小是由用户指定的，即poll的第二个参数。但仍有扫描效率和内存拷贝问题。
pselect模型：I/O多路复用技术。同select。
select和iocp分别对应第3种与第5种模型，那么epoll与kqueue呢？其实也于select属于同一种模型，只是更高级一些，可以看作有了第4种模型的某些特性，如callback机制。
那么，为什么epoll,kqueue比select高级？
答案是，他们无轮询。因为他们用callback取代了。想想看，当套接字比较多的时候，每次select()都要通过遍历FD_SETSIZE个Socket来完成调度,不管哪个Socket是活跃的,都遍历一遍。这会浪费很多CPU时间。如果能给套接字注册某个回调函数，当他们活跃时，自动完成相关操作，那就避免了轮询，这正是epoll与kqueue做的。
windows or *nix （IOCP or kqueue/epoll）？
诚然，Windows的IOCP非常出色，目前很少有支持asynchronous I/O的系统，但是由于其系统本身的局限性，大型服务器还是在UNIX下。而且正如上面所述，kqueue/epoll 与 IOCP相比，就是多了一层从内核copy数据到应用层的阻塞，从而不能算作asynchronous I/O类。但是，这层小小的阻塞无足轻重，kqueue与epoll已经做得很优秀了。
提供一致的接口，IO Design Patterns
实际上，不管是哪种模型，都可以抽象一层出来，提供一致的接口，广为人知的有ACE,Libevent这些，他们都是跨平台的，而且他们自动选择最优的I/O复用机制，用户只需调用接口即可。说到这里又得说说2个设计模式，Reactor and Proactor。有一篇经典文章值得阅读，Libevent是Reactor模型，ACE提供Proactor模型。实际都是对各种I/O复用机制的封装。
Java nio包是什么I/O机制？
我曾天真的认为java nio封装的是IOCP。。现在可以确定，目前的java本质是select()模型，可以检查/jre/bin/nio.dll得知。至于java服务器为什么效率还不错。。我也不得而知，可能是设计得比较好吧。。-_-。
=====================分割线==================================
总结一些重点：
- 只有IOCP是asynchronous I/O，其他机制或多或少都会有一点阻塞。
- select低效是因为每次它都需要轮询。但低效也是相对的，视情况而定，也可通过良好的设计改善
- epoll, kqueue是Reacor模式，IOCP是Proactor模式。
- java nio包是select模型。