Linux下编写高性能的网络程序

最简单的tcp服务器端程序

我们先来实现一个最简单的tcp服务器程序。

#include <iostream>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>

using namespace std;

int main() {
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if(sockfd == -1) {
                perror("socket");
                exit(EXIT_FAILURE);
        }
        sockaddr_in address;
        address.sin_family = AF_INET;
        address.sin_addr.s_addr = htonl(INADDR_ANY);
        address.sin_port = htons(9090);
        int addr_len = sizeof(address);


        int res = bind(sockfd, (struct sockaddr*)&address, addr_len);
        listen(sockfd, 5);
        if(res == -1) {
                perror("bind");
                exit(EXIT_FAILURE);
        }

        const char *msg = "Hello. I'm server.";
        char data_rcv[1024];
        sockaddr_in client_addr;
        int client_fd;
        socklen_t client_socklen = sizeof(client_addr);

        while(true) {
                cout<<"server is waiting for connect..."<<endl;
                client_fd = accept(sockfd, (struct sockaddr*)&client_addr, &client_socklen);
                if(client_fd == -1) {
                        perror("accept");
                        exit(EXIT_FAILURE);
                }
                int nwrite = write(client_fd, msg, strlen(msg));
                cout<<"write "<<nwrite<<" bytes"<<endl;
                close(client_fd);
        }


        close(sockfd);


        return EXIT_SUCCESS;
}

这个服务器程序简单地建立、绑定、监听套接字，然后逐一地accept来自客户端的请求。这是最简单的模型，只能串行地先处理一个请求，在对客户端write完之后，关闭客户端套接字，然后再accept下一个请求。在网络不好的情况下，在当前请求的处理中，由于write等函数可能被网络阻塞，导致执行异常缓慢，下一个请求可能被block很久。毫无疑问，这样的服务器模型是肯定不能满足需要应付大访问量的业务需求的。需要注意的是一些数据成员我们要先通过hton系列函数把它转换成网络字节序（网络字节序是大端表示，我们的PC大多数都是小端机器），否则会出现奇怪的现象，比如说没有把端口号9090通过htons转换成网络字节序的话，通过netstat -tln查看发现打开的并不是9090端口，而是33315端口（因为9090的小端表示是0x8223，在大端看来，0x8223等于十进制的33315）。下图是9090没有通过htons转换的结果：

telnet可以帮我们测试这个小程序。图中红框部分是server的响应消息。在发送完消息之后，服务器关闭了连接。

将其改造成http服务器

为了方便压力测试，我们将其改造成简单的http echo服务器，先定义对应的响应头和相应体，并初始化相应头。

char *header_pattern = "HTTP/1.0 200 OK\r\nContent-length: %d\r\nContent-type: text/html\r\n\r\n";
char header[512];
char *body = "hello, world";
sprintf(header, header_pattern, strlen(body));

在header_pattern中，预留了%d存放body的长度，第四行使用sprintf将body的长度写入其中。
然后在accept之后，先后把响应头和响应体写到客户端（注意：是分两次write而不是header和body合起来一起写）。

write(client_fd, header, strlen(header));
write(client_fd, body, strlen(body));

然后我们就可以通过浏览器访问我们的”服务器了“。

异步响应多客户端

要克服后面的请求会被block很久的毛病，我们必须想办法让服务器程序可以”并行“地处理正在排队等待处理的请求。多线程可以帮我们做到这一点。

多线程实现

我们用多线程方法改一下代码，得到以下程序。

#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <stdio.h>

using namespace std;

const char *content = "hello,world";
const char *header = "HTTP/1.0 200 OK\r\nContent-length: 11\r\nContent-type: text/html\r\n\r\n";


void* client_handler(void *args) {
        int cli_fd = (int)(long)args;
        char *buf = new char[BUFSIZ];
        read(cli_fd, buf, BUFSIZ);
        delete []buf;
        write(cli_fd, header, strlen(header));
        write(cli_fd, content, strlen(content));
        close(cli_fd);
        pthread_exit(NULL);

}

int main() {
        int sock_fd = socket(AF_INET, SOCK_STREAM, 0);
        sockaddr_in serv_addr;
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
        serv_addr.sin_port = htons(9090);
        socklen_t serv_addr_len = sizeof(serv_addr);

        if(bind(sock_fd, (sockaddr*)&serv_addr, serv_addr_len) == -1) {
                perror("bind");
                exit(EXIT_FAILURE);
        }

        if(listen(sock_fd, 128) != 0) {
                perror("listen");
                exit(EXIT_FAILURE);
        }

        cout<<"start listening at port 9090 ..."<<endl;


        pthread_t tid;
        pthread_attr_t attr;
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

        int cli_fd;
        sockaddr_in cli_addr;
        socklen_t cli_addr_len = sizeof(cli_addr);

        while(true) {
                cli_fd = accept(sock_fd, (struct sockaddr*)&cli_addr, &cli_addr_len);
                if(cli_fd == -1) {
                        perror("accept");
                        exit(EXIT_FAILURE);
                }
                pthread_create(&tid, &attr, client_handler, (void*)(long)cli_fd);
        }


        close(sock_fd);

        return 0;
}

思路很简单，每次accept了一个连接之后，启动单独启动一个额外的线程（设置其属性为脱离线程，这样我们就不用在主线程去join 了），由这个额外的线程去负责读写刚才accept的客户端套接字。由于主线程启动一个额外的线程之后可以立刻返回，进而可以继续accept，等待下一个连接的到来，下一个连接到来之后继续启动线程来处理。使用线程的好处是明显的，这样下一个连接就不用等待上一个连接完可以被服务器accept，并与其它请求“并行”地被服务器处理。但是多线程服务器模型有一个问题，那就是当来自客户端的请求量很大的时候，需要启动很多线程来负责处理请求，由于存在非常多的线程，系统仅仅调度这些线程便消耗了大量的的资源，所以此模型并不适合高访问量的场景。

抛弃线程，使用多路复用函数select

select函数可以让我们编写更加高性能的网络程序。我们可以把accept来的一堆客户端套接字放到select里面的套接字集，然后阻塞在select函数中，一旦有套接字可读可写，select会立刻返回，我们处理完相应的工作后（读写完可读可写的套接字之后），继续在select中阻塞，直到再次有请求的到来。这样就免去了系统调度一大堆线程的开支。我们用一个set来保存已有的套接字（源代码在下面给出），每一次都把set中的最大值加一放到select的第一个参数里面，下面的for循环进行套接字检查（从0开始到套接字的最大值遍历，用FD_ISINSET来检查当前套接字是否在套接字集合里面）的时候也直接用这个最大值作为上限，这样我们就不用每次都检查FD_SIZE那么多次了。

使用非阻塞

在select返回之后，便开始对可读可写的套接字进行读写，然后接着阻塞在select中等待下一次有套接字可读可写。这样做有一个小问题，在select返回之后，在对相应套接字的读写过程中，可能会出现网络不好的现象，这样的话一些write和read函数就会阻塞，排在后面的可读可写的套接字就会一直得不到处理。
我们可以用非阻塞读写来解决这个问题。我们把所有的套接字都用fcntl来设置成非阻塞。

int flag = fcntl(sock_fd, F_GETFL, 0);
flag |= O_NONBLOCK;
fcntl(sock_fd, F_SETFL, flag);

这种情况下，无论读写成功或者失败，或者只读或者只写了一部分，read和write函数会立刻返回，这样的话就不会影响后面的请求的处理了。因为write在非阻塞的情况下是遵守“能写多少就写多少的情况”，所以需要额外的数据结构来保存对每一个套接字已经写了多少（保存写的进度），以程序方便select下次返回的时候对相应的套接字在上次的断点上继续写数据。具体实现如下：

#include <iostream>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <string.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <set>


using namespace std;

#define INIT_PROGRESS 100000000


//pair的比较类，这样做是为了确保pair.first为key（在set中不会出现重复）。
struct CompPair {
        bool operator ()(const pair<int, size_t> &p1, const pair<int, size_t> &p2) const {
                return p1.first < p2.first;
        }
};


//读取文件内容并返回
char* read_file_data(const char *file_name, size_t *len) {
        int fd = open(file_name, O_RDONLY);
        struct stat st;
        char *result = NULL;
        if(fd != -1 && fstat(fd, &st) != -1) {
                *len = st.st_size;
                result = new char[*len + 1];
                int nread = read(fd, result, *len);
                if(*len != nread) {
                        delete []result;
                        result = NULL;
                }
        }

        close(fd);
        return result;
}


//释放掉存放文件内容的指针
void *free_file_data(char* p) {
        if(p != NULL) {
                delete []p;
        }
}


int main(int argc, char **argv) {
		/* 用法是命令后面直接跟一个文件名字，
		*  这个文件就是要分享的文件，其内容
		*  将作为响应的body，将会出现在浏览
		*  器上，这样做事为了方便后面的性能
		*  测试
		*/
        if(argc != 2) {
                cout<<"usage: "<<argv[0]<<" <file to share>"<<endl;
                exit(EXIT_SUCCESS);
        }

        //读取文件数据，长度存在file_len中
        size_t file_len;
        char *file_data = read_file_data(argv[1], &file_len);
        if(file_data != NULL) {
                cout<<"file size: "<<file_len<<endl;
        }
        else {
                cout<<"open file error"<<endl;
                exit(EXIT_FAILURE);
        }

        const char *header_pattern = "HTTP/1.0 200 OK\r\nContent-length: %d\r\nContent-type: text/html\r\n\r\n";
        char header[1024];
        //把body的长度写入header当中
        sprintf(header, header_pattern, file_len);
        int header_len = strlen(header);

        //保存进度的set，pair.first是相应的套接字值，第二个参数是已经写了的字节数
        //负数表示正在写header，正数表示正在写body
        set<pair<int, size_t>, CompPair> progress_set;


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

        //保存全部套接字的set，这样做事为了获取套接字最大值再加一传给select的第一个参数
        set<int> sock_fd_set;
        sock_fd_set.insert(sock_fd + 1);

        //把监听套接字设成非阻塞的
        int flag = fcntl(sock_fd, F_GETFL, 0);
        flag |= O_NONBLOCK;
        fcntl(sock_fd, F_SETFL, flag);

        sockaddr_in serv_addr, cli_addr;
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_port = htons(9734);
        serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
        socklen_t cli_addr_len = sizeof(cli_addr);
        int serv_addr_len = sizeof(serv_addr);
        //绑定
        int res = bind(sock_fd, (sockaddr*)&serv_addr, serv_addr_len);
        if(res == -1) {
                perror("bind");
                exit(EXIT_FAILURE);
        }
        //监听
        listen(sock_fd, 128);

        fd_set read_set, tmp_set, write_set, tmp_set1;
        FD_ZERO(&read_set);
        FD_ZERO(&write_set);
        FD_SET(sock_fd, &read_set);
        char buf[BUFSIZ];


        while(true) {
                tmp_set = read_set;
                tmp_set1 = write_set;
                timeval timeout = { .tv_sec = 5, .tv_usec = 500000 };

                //select函数调用，第一个参数是套接字的最大值加一，后面分别是需要读的套接字集，需要写的套接字集，错误输出套接字集
                int select_result = select(*(--sock_fd_set.end()), &tmp_set, (fd_set*)&tmp_set1, (fd_set*)NULL, &timeout);

                if(select_result == 0) {
                        cout<<"timeout"<<endl;
                }
                else if(select_result == -1) {
                        perror("select");
                        exit(EXIT_FAILURE);
                }
                else {
                        int i;
                        //获取套接字最大值
                        int max_sock_fd = *(--sock_fd_set.end());
                        for(i = 0; i < max_sock_fd; i++) {
                        	//检查是否在可写套接字集中
                                if(FD_ISSET(i, &tmp_set1)) {
                                        int start_pos = 0;
                                        //找到当前套接字的写进度
                                        set<pair<int, size_t> >::iterator iter = progress_set.find(pair<int, size_t>(i, 0));
                                        if(iter != progress_set.end()) {
                                                start_pos = iter->second;
                                        }
                                        else {
                                                cout<<"progress error"<<endl;
                                                exit(EXIT_FAILURE);
                                        }




                                        pair<int, size_t> pair_to_insert(iter->first, 0);

                                        progress_set.erase(iter);

                                        //如果进度是INIT_PROGRESS，则开始写头部
                                        if(start_pos == INIT_PROGRESS) {
                                                //cout<<"begin write header"<<endl;
                                                int nwrite = write(i, header, header_len);
                                                if(nwrite == header_len) {
                                                        pair_to_insert.second = 0;
                                                        //cout<<"finish write header"<<endl;
                                                }
                                                else {
                                                        start_pos = -1 - nwrite;
                                                        pair_to_insert.second = start_pos;
                                                }
                                        }
                                        //如果进度大于等于0，写响应体
                                        else if(start_pos >= 0) {
                                                //cout<<"write body"<<endl;
                                                int bytes_to_write = file_len - start_pos;
                                                int nwrite = write(i, file_data + start_pos, bytes_to_write);
                                                if(bytes_to_write == nwrite) {
                                                        FD_CLR(i, &write_set);
                                                        FD_CLR(i, &read_set);
                                                        close(i);
                                                        //cout<<"send to fd "<<i<<" done"<<endl;
                                                        continue;
                                                }
                                                else {
                                                        pair_to_insert.second = start_pos + nwrite;
                                                }
                                        }
                                        //如果进度为负，继续写头部，实际进度为进度的绝对值再减一
                                        else {
                                                //cout<<"write header"<<endl;
                                                start_pos = -start_pos - 1;
                                                int bytes_to_write = header_len - start_pos;
                                                int nwrite = write(i , header + start_pos, bytes_to_write);
                                                //写完了头部之后把进度设成0，告诉程序下次应该开始写响应体了
                                                if(nwrite == bytes_to_write) {

                                                        pair_to_insert.second = 0;
                                                }
                                                //否则保存好进度
                                                else {
                                                        pair_to_insert.second = -1 - start_pos - nwrite;
                                                }
                                        }

                                        progress_set.insert(pair_to_insert);
                                }
                                //检查是否在可读套接字集中
                                if(FD_ISSET(i, &tmp_set)) {
                        		//如果是监听套接字，则accept新的客户端套接字，然后把新的客户端
                        		//套接字加入到读和写套接字集中
                                        if(i == sock_fd) {
                                                int cli_fd = accept(i, (sockaddr*)&cli_addr, &cli_addr_len);
                                                if(cli_fd == -1) {
                                                        perror("accept");
                                                        continue;
                                                }


                                                sock_fd_set.insert(cli_fd + 1);
                                                //初始化进度为INIT_PROGRESS
                                                progress_set.insert(pair<int, size_t>(cli_fd, INIT_PROGRESS));

                                                //设置成非阻塞
                                                flag = fcntl(cli_fd, F_GETFL, 0);
                                                flag |= O_NONBLOCK;
                                                fcntl(cli_fd, F_SETFL, flag);

                                                //加入到需要读的套接字集中
                                                FD_SET(cli_fd, &read_set);
                                        }

                                        else {
                                        		//如果不是监听套接字，就读取请求头
                                                int nread = read(i, buf, BUFSIZ);
                                                if(nread == 0) {
                                                        close(i);
                                                        FD_CLR(i, &read_set);
                                                        FD_CLR(i, &write_set);

                                                        set<int>::iterator iter = sock_fd_set.find(i + 1);
                                                        if(iter != sock_fd_set.end()) {
                                                                sock_fd_set.erase(iter);
                                                                cout<<"max sock is "<<*(--sock_fd_set.end())<<endl;;
                                                        }
                                                        else {
                                                                cout<<"set error"<<endl;
                                                        }

                                                        cout<<"fd "<<i<<" closed"<<endl;
                                                        continue;
                                                }
                                                if(nread == -1) {
                                                        perror("read");
                                                        continue;
                                                }
                                                
                                                //读取成功后，把套接字加入到需要写的套接字集中，为写响应头做准备
                                                FD_SET(i, &write_set);
                                        }

                                }


                        }


                }
        }


        free_file_data(file_data);

        return 0;
}

使用更高性能的epoll代替select

epoll是linux2.5.44内核之后的产物，具有更高的性能。使用方法跟select大同小异，这里就不阐述了。

性能测试

以上程序性能的高低都还是理论上的，还没有进行相应的实际测试。我们使用ab压力测试工具来对以上几个程序进行性能测试。

可见，epoll的性能是最高的，select略逊色于epoll，而多线程性能是最低的。