[原创]教你开发一个高效的服务器

asert · 发表于 2008-7-28 18:02:17

教你开发一个高效的服务器

转载请注明作者信息
作者: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;
}
}
}
}

复制代码

realtang · 发表于 2008-7-29 09:13:03

asert兄自己的流媒体服务器也是采用这个框架？
如果这个框架还能解决多进程的问题，类似apache那样，就更好了。

asert · 发表于 2008-7-29 10:00:58

Post by realtang;1879672
asert兄自己的流媒体服务器也是采用这个框架？
如果这个框架还能解决多进程的问题，类似apache那样，就更好了。

这个框架是我这段时间一直做各种服务器开发总结出来的经验，以后要开发的流媒体服务器也会用这个框架的，不过现在还没有做流媒体服务器，正在学习av方面的知识。

你说的多进程的问题能再详细一点说说是什么问题吗？我觉得我的这个框架写的还不错，如果知道能有还需要改进的地方那我就又有趣的事可做了，呵呵

xiaolin3325 · 发表于 2008-9-27 15:35:39

收藏，改天慢慢看!!

		自动登录	找回密码
密码			注册