|
教你开发一个高效的服务器
转载请注明作者信息
作者:lijiuwei
邮箱:lijiuwei0902@gmail.com
高效服务器的关键技术:
1.需要有固定数量的线程来去处理请求,而不可以每次收到请求都fork或者create_thread;
2.如果socket使用堵塞模式,那么read必须要有超时,如果使用非堵塞模式,那么read就必须要有数据缓冲;
3.使用epoll分别在不同的线程中监视可读和可写;
4.对于tcp长连接使用主线程分配连接,然后由任务线程处理,一个任务线程对应一个任务队列;对于tcp短连接和udp连接,使用竞争线程;
5.时刻检测并且关闭超时连接;
6.真正关闭socket的地方只能由一处,其他地方发现其socket需要关闭只能标识socket应该关闭了,然后由那一处安全的关闭socket;
7.当服务器需要不断的发送硬盘上的数据给客户端(比如说文件传输服务器和流媒体服务器),那么服务器端必须先从硬盘加载数据到数据缓存,再从数据缓冲发送数据给客户端;
8.内存分配必须成功,不然线程睡眠然后重新分配直到分配成功为止;
通用设计方案:
现在我们设计一个通用的服务器,这个服务器接受客户端的连接,然后不断读取客户端的请求,在某一时刻又转为不断的发送数据给客户端;下面我们看看怎么设计这样的一个服务器
注:
1.为了只关注设计方面并且让代码尽量简洁,对于一些错误处理我删除了。
2.must_malloc,must_calloc和其他前缀为must的函数封装了相应的内存分配函数,使得分配必须成功;
3.queue_t结构是一个通用的任务队列结构,queue_new是创建新的队列,queue_push是推数据到队列中,queue_pop是从队列中抛出数据
4.下面的代码只是一个良好的服务器架构,而不是一个真正能使用的服务器
5.欢迎讨论和优化!
- typedef struct server_st *server_t;
- typedef struct task_queue_st *task_queue_t;
- typedef struct conn_st *conn_t;
- /** 服务器结构 */
- struct server_st {
- /** 监听地址 */
- struct in_addr listen_addr;
- /** 监听端口 */
- int port;
- /** 监听描述符 */
- int listen_fd;
- /**负责读取请求的任务队列,一个线程负责处理一个任务队列,一共五个*/
- queue_t readreq_task_queues[5];
- /**负责写数据的任务队列,一个线程负责处理一个任务队列,一共五个*/
- queue_t writedata_task_queues[5];
- };
- /** 任务队列结构 */
- struct task_queue_st {
- /** 待添加的连接 */
- queue_t pending_conn_queue;
- /** 互斥锁,用于多线程环境下对任务队列的访问控制 */
- pthread_mutex_t task_queue_mutex;
- /** 链表结构,保存任务队列中的所有连接 */
- conn_t conn_head;
- /** 链表的最后一个节点,用于方便在结尾插入新的节点 */
- conn_t conn_tail;
- /** 总连接数*/
- int total_conns;
- };
- /** 用户的阶段状态 */
- typedef enum {
- status_ESTABLISHED,/** 客户端刚建立连接 */
- status_CONNECTED,/** 客户端正在连接 */
- status_START_RECV_DATA,/** 客户端开始接收数据,即表示服务器端开始发送数据 */
- status_CLOSED/** 客户端关闭连接 */
- } client_status_t;
- /** 服务器的阶段状态 */
- typedef enum {
- status_SERVER_CONNECTED,/** 与客户端建立连接 */
- status_SERVER_CLOSING /** 服务器关闭连接 */
- } server_status_t;
- /** 连接结构 */
- struct conn_st {
- /** 该连接的所属服务器 */
- server_t server;
- /** 该连接的引用计数 */
- int ref_count;
- /** 用户最后一次活动的时间*/
- time_t last_activity;
- /** 客户端当前的状态阶段*/
- client_status_t client_status;
- /** 服务器当前的状态阶段*/
- server_status_t server_status;
- /** socket描述符*/
- int fd;
- /** 数据缓冲,当服务器端需要不断的发送数据给客户端时被用到 */
- char *remain_data;
- /** 数据缓冲剩余数据大小 */
- int remain_data_size;
- /** 下一个连接 */
- conn_t readreq_task_queue_next;
- /** 下一个连接 */
- conn_t writedata_task_queue_next;
- };
- /** 主函数 */
- int main(int argc, char **argv) {
- int idx, conn_fd;
- server_t server;
- server = must_calloc(1, sizeof(struct server_st));
- server->port = 端口号;
- server->listen_addrs.s_addr = INADDR_ANY;
- server->listen_fd = 服务器监听(&server->listen_addr,server->port);
- 设置成非堵塞(server->listen_fd);
- /* 创建5个负责读取请求的任务线程 */
- for (idx = 0; idx < 5; idx++) {
- task_queue_t task_queue = server->readreq_task_queues[idx];
- bzero(task_queue,sizeof(struct task_queue_st));
- task_queue->pending_conn_queue = queue_new();
- pthread_mutex_init(&task_queue->task_queue_mutex,NULL);
- pthread_create(&task_queue->thread_tid, NULL, readreq_task_thread, task_queue);
- }
- /* 创建5个负责写数据的任务线程 */
- for (idx = 0; idx < 5; idx++) {
- task_queue_t task_queue = server->writedata_task_queues[idx];
- bzero(task_queue,sizeof(struct task_queue_st));
- task_queue->pending_conn_queue = queue_new();
- pthread_mutex_init(&task_queue->task_queue_mutex,NULL);
- pthread_create(&task_queue->thread_tid, NULL, writedata_task_thread, task_queue);
- }
- /* 无限循环的监听,等待客户端的连接请求 */
- while (1) {
- conn_fd = accept(server->listen_fd, 0, 0);
- if (conn_fd < 0) continue;
- /* 和一个客户端建立了连接,下面开始把该连接分配给负责读取请求的任务线程*/
- assign_to_readreq_task_thread(server,conn_fd);
- }
- return 0;
- }
- /*分配任务,原则为哪个线程的任务队列最少就分配给哪一个线程去处理*/
- static void assign_to_readreq_task_thread(server_t server, int conn_fd) {
- int idx;
- task_queue_t task_queue, min_task_queue = NULL;
- conn_t conn;
- /* 查找连接数最少的任务队列 */
- for (idx = 0; idx < 5; idx++) {
- task_queue = server->readreq_task_queues[idx];
- if (min_task_queue == NULL)
- min_task_queue = task_queue;
- else if (min_task_queue->total_conns > task_queue->total_conns)
- min_task_queue = task_queue;
- }
- if (min_task_queue) {
- conn = must_calloc(1, sizeof(struct conn_st));
- conn->server = server;
- conn->client_status = status_ESTABLISHED;
- conn->server_status = status_SERVER_CONNECTED;
- conn->ref_count++;/*累加引用计数*/
- /*添加到该任务队列中的待添加队列并且累加该任务队列里的连接数*/
- pthread_mutex_lock(&min_task_queue->task_queue_mutex);
- queue_push(min_task_queue->pending_conn_queue, conn, 1);
- min_task_queue->total_conns++;
- pthread_mutex_unlock(&min_task_queue->task_queue_mutex);
- }
- }
- /**
- *负责读取请求的任务线程处理主函数,负责管理,处理自己队列中的所有连接
- */
- static void *readreq_task_thread(void *arg) {
- int epfd = epoll_create(20480);
- struct epoll_event ev,happened_ev[128];
- task_queue_t task_queue = (task_queue_t) arg;
- int ready,i;
- while (1) {
- time_t curtime = time(NULL);
- /*下面三个变量遍历链表时使用,conn是当前连接,prev_conn是先前的连接,discarded_conn是丢弃的连接*/
- conn_t conn = NULL, prev_conn = NULL, discarded_conn;
- /*把待添加队列里的连接添加到任务队列中*/
- pthread_mutex_lock(&task_queue->task_queue_mutex);
- while ((conn = queue_pop(task_queue->pending_conn_queue))) {
- if (!task_queue->conn_head) {
- task_queue->conn_head = conn;
- } else {
- task_queue->conn_tail->readreq_task_queue_next = conn;
- }
- task_queue->conn_tail = conn;
- }
- pthread_mutex_unlock(&task_queue->task_queue_mutex);
- if (task_queue->total_conns == 0) {
- thread_sleep(1);
- continue;
- }
- conn = task_queue->conn_head;
- while (conn) {
- //清理已经关闭的连接,把已经关闭的连接和超时的连接从链表中删除掉
- boolean conn_timeout = conn->last_activity > 0 && curtime - conn->last_activity > CLIENT_TIMEOUT;
- if ((conn->client_status == status_CLOSED || conn->server_status == status_SERVER_CLOSING || conn_timeout) && conn->ref_count == 1) {
- task_queue->total_conns--;
- discarded_conn = conn;//先放入临时变量中
- //从链表中删除
- if (prev_conn) {
- if (!conn->readreq_task_queue_next) {
- task_queue->conn_tail = prev_conn;
- prev_conn->readreq_task_queue_next = NULL;
- } else {
- prev_conn->readreq_task_queue_next = conn->readreq_task_queue_next;
- }
- } else {
- task_queue->conn_head = conn->readreq_task_queue_next;
- }
- conn = conn->readreq_task_queue_next;
- close(discarded_conn->fd);/*真正关闭连接的地方*/
- free(discarded_conn);
- } else {
- ev.data.fd = conn->fd;
- ev.data.ptr = conn;
- /*只负责监视读事件和出错事件*/
- ev.events = EPOLLIN | EPOLLET;
- if(conn->client_status == status_ESTABLISHED) {
- 设置超时时间或者设置非堵塞(conn->fd);/*两者必须选一个*/
- epoll_ctl(epfd,EPOLL_CTL_ADD,conn->fd,&ev);
- conn->client_status = status_CONNECTED;
- } else {
- epoll_ctl(epfd,EPOLL_CTL_MOD,conn->fd,&ev);
- }
- prev_conn = conn;
- conn = conn->readreq_task_queue_next;
- }
- }
- /** 堵塞等待,超时时间为POLL_TIMEOUT毫秒*/
- if ((ready = epoll_wait(epfd, happened_ev,sizeof(happened_ev) / sizeof(struct epoll_event), POLL_TIMEOUT)) < 0)
- continue;
- /**
- * 循环查找向我们发送请求的客户端的连接
- */
- for (i = 0; i < ready; i++) {
- if (happened_ev[i].events & POLLIN) {
- conn = (conn_t) happened_ev[i].data.ptr;
- conn->last_activity = time(NULL);/* 记录客户端的最后活动时间*/
- proc_protocol(conn);/*处理协议函数*/
- } else if(happened_ev[i].events & EPOLLHUP) {
- conn = (conn_t) happened_ev[i].data.ptr;
- conn->client_status = status_CLOSED;
- }
- }
- }
- }
- /* 协议处理函数 */
- void proc_protocol(conn_t conn) {
- 某一时刻客户端要求服务器端不断发送数据给客户端,然后设置了conn->client_status = status_START_RECV_DATA,并且调用了下面的move_to_writedata_task_thread(conn);
- }
- /*分配任务,原则为哪个线程的任务队列最少就分配给哪一个线程去处理*/
- static void move_to_writedata_task_thread(conn_t conn) {
- server_t server = conn->server;
- task_queue_t task_queue, min_task_queue= NULL;
- int idx;
- /* 查找连接数最少的任务队列 */
- for (idx = 0; idx < 5; idx++) {
- task_queue = server->writedata_task_queues[idx];
- if (min_task_queue == NULL)
- min_task_queue = task_queue;
- else if (min_task_queue->total_conns > task_queue->total_conns)
- min_task_queue = task_queue;
- }
- if (min_task_queue) {
- conn->ref_count++;/*累加引用计数*/
- /*添加到该任务队列中的待添加队列并且累加该任务队列里的连接数*/
- pthread_mutex_lock(&min_task_queue->task_queue_mutex);
- queue_push(min_task_queue->pending_conn_queue, conn, 1);
- min_task_queue->total_conns++;
- pthread_mutex_unlock(&min_task_queue->task_queue_mutex);
- }
- }
- /**
- *负责写数据的线程处理主函数,负责管理,处理自己队列中的所有连接
- */
- static void *writedata_task_thread(void *arg) {
- task_queue_t task_queue = (task_queue_t) arg;
- int epfd = epoll_create(20480);
- struct epoll_event ev,happened_ev[128];
- int ready,i,nwrite,nread,sndbuf_size = 10240;
- char buf[sndbuf_size];
- while (1) {
- time_t curtime = time(NULL);
- /*下面三个变量遍历链表时使用,conn是当前连接,prev_conn是先前的连接,discarded_conn是丢弃的连接*/
- conn_t conn = NULL, prev_conn = NULL, discarded_conn;
- /*把待添加队列里的连接添加到任务队列中*/
- pthread_mutex_lock(&task_queue->task_queue_mutex);
- while ((conn = queue_pop(task_queue->pending_conn_queue))) {
- if (!task_queue->conn_head) {
- task_queue->conn_head = conn;
- } else {
- task_queue->conn_tail->writedata_task_queue_next = conn;
- }
- task_queue->conn_tail = conn;
- }
- pthread_mutex_unlock(&task_queue->task_queue_mutex);
- if (task_queue->total_conns == 0) {
- thread_sleep(1);
- continue;
- }
- conn = task_queue->conn_head;
- while (conn) {
- //清理已经关闭的连接,把已经关闭的连接和超时的连接从链表中删除掉
- boolean conn_timeout = conn->last_activity > 0 && curtime - conn->last_activity > CLIENT_TIMEOUT;
- if (conn->client_status == status_CLOSED || conn->server_status == status_SERVER_CLOSING || conn_timeout) {
- task_queue->total_conns--;
- discarded_conn = conn;//放入临时变量中
- //从链表中删除
- if (prev_conn) {
- if (!conn->writedata_task_queue_next) {
- task_queue->conn_tail = prev_conn;
- prev_conn->writedata_task_queue_next = NULL;
- } else {
- prev_conn->writedata_task_queue_next = conn->writedata_task_queue_next;
- }
- } else {
- task_queue->conn_head = conn->writedata_task_queue_next;
- }
- conn = conn->writedata_task_queue_next;
- discarded_conn->ref_count--;/*引用计数减一,并且该连接已经关闭了,那么真正关闭连接的任务就交给readreq_task_thread函数了*/
- } else {
- ev.data.fd = conn->fd;
- ev.data.ptr = conn;
- /*只负责监视写事件和出错事件*/
- ev.events = EPOLLOUT | EPOLLET;
- if(conn->client_status == status_START_RECV_DATA) {
- setsockopt(conn->fd, SOL_SOCKET, SO_SNDBUF, (char *)&sndbuf_size, sizeof(int));
- epoll_ctl(epfd,EPOLL_CTL_ADD,conn->fd,&ev);
- conn->client_status = status_CONNECTED;
- } else {
- epoll_ctl(epfd,EPOLL_CTL_MOD,conn->fd,&ev);
- }
- prev_conn = conn;
- conn = conn->writedata_task_queue_next;
- }
- }
- /**堵塞等待,超时时间为POLL_TIMEOUT毫秒*/
- if ((ready = epoll_wait(epfd, happened_ev,sizeof(happened_ev) / sizeof(struct epoll_event), POLL_TIMEOUT)) < 0)
- continue;
- /**
- * 循环查找向我们发送请求的客户端的连接
- */
- for (i = 0; i < ready; i++) {
- if (happened_ev[i].events & EPOLLOUT) {
- conn = (conn_t) happened_ev[i].data.ptr;
- conn->last_activity = time(NULL);/* 记录客户端的最后活动时间*/
- if(conn->remain_data_size == 0) {/*如果数据缓存已经没有任何数据了的话*/
- if((nread = read(文件描述符,buf,sizeof(buf))) > 0) {/*从硬盘读数据出来到数据缓存*/
- if((nwrite = write(conn->fd,buf,nread)) < 0 && errno != EINTR) {
- conn->client_status = status_CLOSED;
- continue;
- }
- if(nwrite < nread) {
- conn->remain_data_size = nread - nwrite;
- conn->remain_data = must_malloc(conn->remain_data_size);
- memcpy(conn->remain_data,buf + nwrite, conn->remain_data_size);
- }
- } else if(nread == 0) {
- conn->server_status = status_SERVER_CLOSING;
- }
- } else {
- /*写数据缓存里的数据到客户端*/
- if((nwrite = write(conn->fd,conn->remain_data,conn->remain_data_size)) < 0 && errno != EINTR) {
- conn->client_status = status_CLOSED;
- free(conn->remain_data);
- continue;
- }
- if(nwrite < conn->remain_data_size) {
- conn->remain_data_size = conn->remain_data_size - nwrite;
- memcpy(buf, conn->remain_data + nwrite, conn->remain_data_size);
- conn->remain_data = must_realloc(conn->remain_data, conn->remain_data_size);
- memcpy(conn->remain_data, buf, conn->remain_data_size);
- } else {
- free(conn->remain_data);
- conn->remain_data = NULL;
- conn->remain_data_size = 0;
- }
- }
- } else if(happened_ev[i].events & EPOLLHUP) {
- conn = (conn_t) happened_ev[i].data.ptr;
- conn->client_status = status_CLOSED;
- }
- }
- }
- }
复制代码 |
|