EPOLL原理详解

struct sock {

    __u32   daddr;  /* 外部IP地址   */

    __u32   rcv_saddr; /* 绑定的本地IP地址  */

    __u16   dport;  /* 目标端口   */

    __u16   sport;  /* 源端口    */

    unsigned short  family;  /* 地址簇   */

    int   rcvbuf;  /* 接受缓冲区长度(单位：字节) */

    struct sk_buff_head receive_queue; /* 接收包队列   */

    int   sndbuf;  /* 发送缓冲区长度(单位：字节)  */

    struct sk_buff_head write_queue; /* 包发送队列   */

    wait_queue_head_t *sleep;  /* 等待队列，通常指向socket的wait域 */

    ......

}

public static void main(String[] args) throws IOException {

    ServerSocket serverSocket = new ServerSocket(9000);

    while (true) {

        // 没有连接-阻塞

        Socket socket = serverSocket.accept();

        byte[] bytes = new byte[1024];

        InputStream inputStream = socket.getInputStream();

        while (true) {

            // 没有数据-阻塞

            int read = inputStream.read(bytes);

            if (read != -1) {

                System.out.println(new String(bytes, 0, read));

            } else {

                break;

            }

        }

        socket.close();

    }

}

int s = socket(AF_INET, SOCK_STREAM, 0);  

bind(s, ...)

listen(s, ...)

//接受客户端连接

int c = accept(s, ...)

int readSet[] =  存放需要监听的socket文件描述符

while(1){

    int n = select(..., readSet, ...)

    for(int i=0; i < readSet.count; i++){

        if(FD_ISSET(readSet[i], ...)){

            //fds[i]的数据处理

        }

    }

}

select 系统调用函数

# maxfd：表示的是待测试的描述符基数，它的值是待测试的最大描述符加 1

# readset：读描述符集合

# writeset：写描述符集合

# exceptset：错误描述符集合

# timeout：超时时间

int select(int maxfd, fd_set *readset, fd_set *writeset, fd_set *exceptset, const struct timeval *timeout);

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

# fd：描述符 fd

# events：描述符上待检测的事件类型events，注意这里的 events 可以表示多个不同的事件，

#         具体的实现可以通过使用二进制掩码位操作来完成，例如，POLLIN 和 POLLOUT 可以表示读和写事件

# revents：

struct pollfd {

    int    fd;       /* file descriptor */    

    short  events;   /* events to look for */    

    short  revents;  /* events returned */ 

};

# 创建了一个 epoll 实例 epollevent

int epoll_create(int size);

# 往这个 epoll 实例增加或删除监控的事件

# epfd：epoll_create 创建的 epoll 实例句柄

# op：增加还是删除一个监控事件

# fd：注册的事件的文件描述符

# event：注册的事件类型，并且可以在这个结构体里设置用户需要的数据

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

# 类似之前的 poll 和 select 函数，调用者进程被挂起，在等待内核 I/O 事件的分发

# epfd：实例描述字，也就是 epoll 句柄

# events：用户空间需要处理的 I/O 事件，这是一个数组，数组的大小由 epoll_wait 的返回值决定，这个数组的每个元素都是一个需要待处理的 I/O 事件，其中 events 表示具体的事件类型，事件类型取值和 epoll_ctl 可设置的值一样，这个 epoll_event 结构体里的 data 值就是在 epoll_ctl 那里设置的 data，也就是用户空间和内核空间调用时需要的数据

# maxevents：大于 0 的整数，表示 epoll_wait 可以返回的最大事件值

# timeout：阻塞调用的超时值，如果这个值设置为 -1，表示不超时；如果设置为 0 则立即返回，即使没有任何 I/O 事件发生

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

public class ServerConnect {

    private static final int BUF_SIZE = 1024;

    private static final int PORT = 8080;

    private static final int TIMEOUT = 3000;

    public static void main(String[] args) {

        selector();

    }

    public static void selector() {

        Selector selector = null;

        ServerSocketChannel ssc = null;

        try {

            // 打开一个Channel

            ssc = ServerSocketChannel.open();

            // 将Channel绑定端口

            ssc.socket().bind(new InetSocketAddress(PORT));

            // 设置Channel为非阻塞，如果设置为阻塞，其实和BIO差不多了。

            ssc.configureBlocking(false);

            // 打开一个Slectore

            selector = Selector.open();

            // 向selector中注册Channel和感兴趣的事件

            ssc.register(selector, SelectionKey.OP_ACCEPT);

            while (true) {

                // selector监听事件，select会被阻塞，直到selector监听的channel中有事件发生或者超时，会返回一个事件数量

                //TIMEOUT就是超时时间，selector初始化的时候会添加一个用于主动唤醒的pipe，待会源码分析会说

                if (selector.select(TIMEOUT) == 0) {

                    System.out.println("==");

                    continue;

                }

                /**

                 * SelectionKey的组成是selector和Channel
                 * 有事件发生的channel会被包装成selectionKey添加到selector的publicSelectedKeys属性中
                 * publicSelectedKeys是SelectionKey的Set集合
                 *下面这一部分遍历，就是遍历有事件的channel
                 */
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    if (key.isAcceptable()) {
                        handleAccept(key);
                    }
                    if (key.isReadable()) {
                        handleRead(key);
                    }
                    if (key.isWritable() && key.isValid()) {
                        handleWrite(key);
                    }
                    if (key.isConnectable()) {
                        System.out.println("isConnectable = true");
                    }
                    //每次使用完，必须将该SelectionKey移除，否则会一直存储在publicSelectedKeys中
                    //下一次遍历又会重复处理
                    iter.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (selector != null) {
                    selector.close();
                }
                if (ssc != null) {
                    ssc.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel ssChannel = (ServerSocketChannel) key.channel();
        SocketChannel sc = ssChannel.accept();
        sc.configureBlocking(false);
        sc.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(BUF_SIZE));
    }
    public static void handleRead(SelectionKey key) throws IOException {
        SocketChannel sc = (SocketChannel) key.channel();
        ByteBuffer buf = (ByteBuffer) key.attachment();
        long bytesRead = sc.read(buf);
        while (bytesRead > 0) {
            buf.flip();
            while (buf.hasRemaining()) {
                System.out.print((char) buf.get());
            }
            System.out.println();
            buf.clear();
            bytesRead = sc.read(buf);
        }
        if (bytesRead == -1) {
            sc.close();
        }
    }
    public static void handleWrite(SelectionKey key) throws IOException {
        ByteBuffer buf = (ByteBuffer) key.attachment();
        buf.flip();
        SocketChannel sc = (SocketChannel) key.channel();
        while (buf.hasRemaining()) {
            sc.write(buf);
        }
        buf.compact();
    }
}