基于 epoll 的网络高并发模型开发

文章目录

1、概述
2、程序开发遭遇的问题
3、问题解决小结
4、代码
5、nginx 和 本程序的测试数据及对比小结
6、github 完整源码
           

1、概述

本篇贴在主要论述在解藕 nginx 源码 epoll 高并发模型过程及过程所遇到的问题，以供大家参考。
           

2、程序开发遭遇的问题

基于 epoll 的网络高并发模型代码开发出来后，发现压力测试只等到达对 nginx 同类测试速率的 1/3。
           

3、问题解决小结

此程序最早是基于网上的一个 select 程序开发的，后来让我改造成 epoll 模型的并发程序，最后又稍作改造并测试，形成现在的程序，最早的程序出处已经忘记了。其中对于 epoll 的 ET 和 LT 模式的使用一开始存在一些误解，这主要是由于一开始参考的网上的一些例子导致的，最终看了一个兄弟的帖子（参考链接1），针对 ET 和 LT 的正确使用，才解决了开始并发静态业务无法达到 nginx 接近的性能问题，总结一下就是：
    epoll 在监听 socket 监听文件符应该使用 LT 模式，在监听 socket 客户端文件符读写事件时应使用 ET 模式，此为正确的使用方式。网上的一些帖子说 nginx 的高并发得益于 epoll 的高性能并且是在 ET 模式下，其实这是有误解的，查看其源码可以知道实际上 nginx 使用 epoll 时并不是仅仅使用了 ET 模式。这里将这个兄弟的帖子贴出来供大家参考：
    链接1：http://www.cppblog.com/ifeng/archive/2011/09/29/157141.html
           

注意：链接1的帖子提到其遭遇的问题是，并发时有部分请求丢失了，但我遇到的问题是请求数多了，这里还没弄白是怎么回事，也许在代码细节和环境差异导致的问题不同，不过使用其方法是可以解决我遇到的问题的。

4、代码

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/resource.h>

#define MAXBUF 1024
#define MAXEPOLLSIZE 20000

#define EPOLL_HELLO "epoll hello!"

/*
setnonblocking – set  nonblocking mode
*/

/* 定义常量 */
#define HTTP_DEF_PORT        80     /* 连接的缺省端口 */
#define HTTP_BUF_SIZE      1024     /* 缓冲区的大小 */
#define HTTP_FILENAME_LEN   256     /* 文件名长度 */

/* 定义文件类型对应的 Content-Type */
struct doc_type
{
    char *suffix; /* 文件后缀 */
    char *type;   /* Content-Type */
};

struct doc_type file_type[] =
{
    {"html",    "text/html"  },
    {"gif",     "image/gif"  },
    {"jpeg",    "image/jpeg" },
    { NULL,      NULL        }
};

//Connection: Keep-Alive

char *http_res_hdr_tmpl = "HTTP/1.1 200 OK\r\nServer: Du's Server <0.1>\r\n"
    "Accept-Ranges: bytes\r\nContent-Length: %d\r\nConnection: Close\r\n"
    "Content-Type: %s\r\n\r\n";


/* 根据文件后缀查找对应的 Content-Type */
char *http_get_type_by_suffix(const char *suffix)
{
    struct doc_type *type;

    for (type = file_type; type->suffix; type++)
    {
        if (strcmp(type->suffix, suffix) == )
            return type->type;
    }

    return NULL;
}

/* 解析请求行 */
void http_parse_request_cmd(char *buf, int buflen, char *file_name, char *suffix)
{
    int length = ;
    char *begin, *end, *bias;

    /* 查找 URL 的开始位置 */
    begin = strchr(buf, ' ');
    begin += ;

    /* 查找 URL 的结束位置 */
    end = strchr(begin, ' ');
    *end = ;

    bias = strrchr(begin, '/');
    length = end - bias;

    /* 找到文件名的开始位置 */
    if ((*bias == '/') || (*bias == '\\'))
    {
        bias++;
        length--;
    }

    /* 得到文件名 */
    if (length > )
    {
        memcpy(file_name, bias, length);
        file_name[length] = ;

        begin = strchr(file_name, '.');
        if (begin)
            strcpy(suffix, begin + );
    }
}


/* 向客户端发送 HTTP 响应 */
int http_send_response(int soc)
{
    int read_len, file_len, hdr_len, send_len;
    char *type;
    char read_buf[HTTP_BUF_SIZE];
    char http_header[HTTP_BUF_SIZE];
    char file_name[HTTP_FILENAME_LEN] = "index.html", suffix[] = "html";
    FILE *res_file;

    /* 得到文件名和后缀，解析 http 功能去掉，本处功能与高并发性能关系不大，这里仅返回固定 response */
    /*http_parse_request_cmd(buf, buf_len, file_name, suffix);

    res_file = fopen(file_name, "rb+");
    if (res_file == NULL)
    {
        printf("[Web] The file [%s] is not existed\n", file_name);
        return 0;
    }

    fseek(res_file, 0, SEEK_END);
    file_len = ftell(res_file);
    fseek(res_file, 0, SEEK_SET);

    type = http_get_type_by_suffix(suffix);

    if (type == NULL)
    {
        printf("[Web] There is not the related content type\n");
        return 0;
    }*/

    /* 构造 HTTP 首部，并发送 */
    hdr_len = sprintf(http_header, http_res_hdr_tmpl, strlen(EPOLL_HELLO), "text/html");
    send_len = send(soc, http_header, hdr_len, );
    if (send_len < )
    {
        //fclose(res_file);
        printf("Send mes Error!\n");
        return ;
    }

    //do /* 发送固定字符串，非从文件 */
    {
        //read_len = fread(read_buf, sizeof(char), HTTP_BUF_SIZE, res_file);
        read_len = strlen(EPOLL_HELLO);
        memset(read_buf, , HTTP_BUF_SIZE);
        memcpy(read_buf, EPOLL_HELLO, read_len);

        if (read_len > )
        {
            send_len = send(soc, read_buf, read_len, );
            //file_len -= read_len;
        }
    } //while ((read_len > 0) && (file_len > 0));

    //fclose(res_file);

    return ;
}

int setnonblocking(int sockfd)
{
    if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, ) | O_NONBLOCK) == -)
    {
        return -;
    }
    return ;
}

/* 读取 http 请求，仅读出未正常解析 */
int read_http(int new_fd)
{
    char buf[MAXBUF + ];
    int len;
    int result = ;

    bzero(buf, MAXBUF + );
    len = recv(new_fd, buf, MAXBUF, );
    if (len < )
    {
        printf("recv mes failed! mistake code is%d,mistake info is '%s'\n", errno, strerror(errno));
        return -;
    }

    return len;
}

int main( int argc, char* argv[])
{
    int listener, new_fd, kdpfd, nfds, n, ret, curfds;
    socklen_t len;
    struct sockaddr_in my_addr,their_addr;
    unsigned int myport, lisnum;
    struct epoll_event ev;
    struct epoll_event events[MAXEPOLLSIZE];
    struct rlimit rt;
    long http_count = ;
    long fd_count = ;
    int http_count_all = ;
    int socket_count = ;

    myport = ;
    lisnum = ;
    /* 设置最大连接数 */
    rt.rlim_max = rt.rlim_cur = MAXEPOLLSIZE;
    if( setrlimit( RLIMIT_NOFILE, &rt) == - )
    {
        perror("setrlimit");
        exit();
    }

    /* 创建监听 socket */
    if( (listener = socket( PF_INET, SOCK_STREAM, )) == -)
    {
        perror("socket");
        exit();
    }


    int reuse = ;
    setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));

    setnonblocking(listener);
    my_addr.sin_family = PF_INET;
    my_addr.sin_port = htons(myport);
    my_addr.sin_addr.s_addr = INADDR_ANY;
    bzero( &(my_addr.sin_zero), );

    if ( bind( listener, (struct sockaddr*)&my_addr, sizeof(struct sockaddr)) == - )
    {
        perror("bind");
        exit();
    }


    if (listen(listener, lisnum) == -)
    {
        perror("listen");
        exit();
    }

    /* 创建 epoll 对象 */
    kdpfd = epoll_create( MAXEPOLLSIZE );
    len = sizeof( struct sockaddr_in );
    ev.events = EPOLLIN ;//| EPOLLET;
    ev.data.fd = listener;

    if( epoll_ctl( kdpfd, EPOLL_CTL_ADD, listener, &ev) <  )  // rege
    {
        fprintf( stderr, "epoll set insertion error: fd=%d\n", listener );
        return -;
    }
    curfds = ;

    pid_t pid = fork();

    if (pid==)
    {
        while()
        {
            /* 等待事件 */
            nfds = epoll_wait(kdpfd, events, MAXEPOLLSIZE, -);
            if( nfds == - )
            {
                perror("epoll_wait");
                break;
            }

            for (n = ; n < nfds; ++n)
            {
                if(events[n].data.fd == listener)
                {
                    while( (new_fd = accept( listener, (struct sockaddr*)&their_addr, &len )) <  )
                    {
                        perror("accept");
                        continue;
                    }

                    fd_count = fd_count +;

                    setnonblocking(new_fd);
                    ev.events = EPOLLIN | EPOLLET;
                    ev.data.fd = new_fd;
                    if( epoll_ctl( kdpfd, EPOLL_CTL_ADD, new_fd, &ev) <  )
                    {
                        fprintf(stderr, "add socket '%d' to  epoll failed! %s\n",
                                        new_fd, strerror(errno));
                        return -;
                    }
                    curfds ++;
                }
                else if(events[n].events&EPOLLIN)
                {
                    new_fd = events[n].data.fd;
                    ret = read_http(new_fd);
                    http_count = http_count +;
                    if (ret <  && errno != )
                    {
                        epoll_ctl(kdpfd, EPOLL_CTL_DEL, new_fd, &ev);
                        curfds--;
                        close(new_fd);
                    }
                    else
                    {
                        ev.data.fd = new_fd;
                            ev.events=EPOLLOUT|EPOLLET;
                            epoll_ctl(kdpfd,EPOLL_CTL_MOD,new_fd,&ev);//修改标识符，等待下一个循环时发送数据，异步处理的精髓
                    }
                }
                else if(events[n].events&EPOLLOUT)
                {
                    int result;
                    new_fd = events[n].data.fd;
                    result = http_send_response(new_fd);
                    epoll_ctl(kdpfd, EPOLL_CTL_DEL, new_fd, &ev);
                    curfds--;
                    close(new_fd);
                    http_count_all++;
                }
            }
        }
    }
    else
    {
        while()
        {
            sleep();
        }
    }

    close( listener );
    return ;
}
           

最后，这里应用层使用了 http 协议进行测试，其实现相对比较简单，仅读取了 http 请求并返回了固定的内容，因为本文的主要目的是说明 epoll 的高并发特性，应用层未做过多开发，那部分代码大家简单看看就行，请大家注意。
    关键字：epoll、EPOLLET、EPOLLLT、高并发。
           

5、nginx 和 本程序的测试数据及对比小结

5.1、对于 nginx 静态欢迎界面的压力测试数据如下所示：

测试命令：ab -c  -n  "http://192.168.1.203/index.html"
测试数据：
This is ApacheBench, Version  <$Revision:  $>
Copyright  Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/

Benchmarking  (be patient)
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Finished  requests


Server Software:        nginx/
Server Hostname:        
Server Port:            

Document Path:          /index.html
Document Length:         bytes

Concurrency Level:      
Time taken for tests:    seconds
Complete requests:      
Failed requests:        
Total transferred:       bytes
HTML transferred:        bytes
Requests per second:     [#/sec] (mean)
Time per request:        [ms] (mean)
Time per request:        [ms] (mean, across all concurrent requests)
Transfer rate:           [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:                           
Processing:                       
Waiting:                          
Total:                            

Percentage of the requests served within a certain time (ms)
  %      
  %      
  %      
  %      
  %      
  %      
  %     
  %     
 %     (longest request)
           

5.2、本程序的压测数据：

测试命令：ab -c  -n  "http://192.168.1.203:5024/index.html"
测试数据：

This is ApacheBench, Version  <$Revision:  $>
Copyright  Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/

Benchmarking  (be patient)
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Completed  requests
Finished  requests


Server Software:        Du's
Server Hostname:        
Server Port:            

Document Path:          /index.html
Document Length:         bytes

Concurrency Level:      
Time taken for tests:    seconds
Complete requests:      
Failed requests:        
Total transferred:       bytes
HTML transferred:        bytes
Requests per second:     [#/sec] (mean)
Time per request:        [ms] (mean)
Time per request:        [ms] (mean, across all concurrent requests)
Transfer rate:           [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:                           
Processing:                       
Waiting:                          
Total:                            

Percentage of the requests served within a certain time (ms)
  %      
  %      
  %      
  %      
  %     
  %     
  %     
  %     
 %     (longest request)
           

小结：通过上面的数据可以看出，本程序 ab 工具测试下的“Requests per second”指标已和 Nginx 相差无几，接近三万。实际上如果对于监听 socket 文件描述符使用 EPOLL ET 模式下，“Requests per second”指标只能到达一万，这个数据和网上其他错误使用 EPOLL ET 模式的兄弟们的数据一致。

6、github 完整源码

https://github.com/raoping2017/epoll_high_concurrency.git

注意本程序可能还存在 bug，用于生产环境注意测试。