【Networking】(轉)一個非常好的epoll+線程池服務器Demo

(轉)一個非常好的epoll+線程池服務器Demo
1 Reply
轉載自：http://zhangyafeikimi.javaeye.com/blog/285193

[cpp]
/**
張亞霏修改

文件名:epoll_demo.c
編譯: gcc epoll_demo.c -pthread

程序源碼如下(請自行編輯宏定義SERVER_IP為自己的IP):
*/

/*Linux 2.6 x86_64 only*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <time.h>
#include <sys/epoll.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <netdb.h>
#include <pthread.h>
#include <errno.h>

/*線程池的線程數量*/
#define THREAD_MAX 1

/*監聽端口*/
#define LISTEN_PORT 8000

/*
監聽端口的數量,從LISTEN_PORT到LISTEN_PORT+LISTEN_MAX-1
*/
#define LISTEN_MAX 1

#define SERVER_IP "127.0.0.1"

typedef struct
{
char ip4[128];
int port;
int fd;
} listen_info;

//服務器參數
static listen_info s_listens[LISTEN_MAX];

//線程池參數
static unsigned int s_thread_para[THREAD_MAX][8];//線程參數
static pthread_t s_tid[THREAD_MAX];//線程ID
pthread_mutex_t s_mutex[THREAD_MAX];//線程鎖

//私有函數
static int init_thread_pool(void);
static int init_listen4(char *ip4, int port, int max_link);

//線程函數
void* test_server4(unsigned int thread_para[]);

int main(int argc, char *argv[])//客戶端驅動
{
//臨時變量
int i, j, rc;

int sock_listen; //監聽套接字
int sock_cli; //客戶端連接
int listen_index;

int epfd;
int nfds;
struct epoll_event ev;
struct epoll_event events[LISTEN_MAX];

socklen_t addrlen; //地址信息長度
struct sockaddr_in addr4; //IPv4地址結構

//線程池初始化
rc = init_thread_pool();
if (0 != rc)
exit(-1);

//初始化服務監聽
for (i = 0; i < LISTEN_MAX; i++)
{
sprintf(s_listens[i].ip4, "%s", SERVER_IP);
s_listens[i].port = LISTEN_PORT + i;
//創建監聽
rc = init_listen4(s_listens[i].ip4, s_listens[i].port, 64);
if (0 > rc)
{
fprintf(stderr, "無法創建服務器監聽於%s:%d\n", s_listens[i].ip4, s_listens[i].port);
perror(NULL);
exit(-1);
}
s_listens[i].fd = rc;
}

//設置集合
epfd = epoll_create(8192);
for (i = 0; i < LISTEN_MAX; i++)
{
//加入epoll事件集合
ev.events = EPOLLIN;
ev.data.u32 = i;//記錄listen數組下標
if (epoll_ctl(epfd, EPOLL_CTL_ADD, s_listens[i].fd, &ev) < 0)
{
fprintf(stderr, "向epoll集合添加套接字失敗(fd =%d)\r\n", rc);
exit(-1);
}
}

//服務循環
for ( ; ; )
{
//等待epoll事件
nfds = epoll_wait(epfd, events, LISTEN_MAX, -1);
//處理epoll事件
for (i = 0; i < nfds; i++)
{
//接收客戶端連接
listen_index = events[i].data.u32;
sock_listen = s_listens[listen_index].fd;
addrlen = sizeof(struct sockaddr_in);
bzero(&addr4, addrlen);
sock_cli = accept(sock_listen, (struct sockaddr *)&addr4, &addrlen);
if (0 > sock_cli)
{
fprintf(stderr, "接收客戶端連接失敗\n");
continue;
}
//查詢空閑線程對
for (j = 0; j < THREAD_MAX; j++)
{
if (0 == s_thread_para[j][0]) break;
}
if (j >= THREAD_MAX)
{
fprintf(stderr, "線程池已滿, 連接將被放棄\r\n");
shutdown(sock_cli, SHUT_RDWR);
close(sock_cli);
continue;
}
//復制有關參數
s_thread_para[j][0] = 1;//設置活動標志為"活動"
s_thread_para[j][1] = sock_cli;//客戶端連接
s_thread_para[j][2] = listen_index;//服務索引
//線程解鎖
pthread_mutex_unlock(s_mutex + j);
}//end of for(i;;)
}//end of for(;;)

exit(0);
}

static int init_thread_pool(void)
{
int i, rc;

//初始化線程池參數
for (i = 0; i < THREAD_MAX; i++)
{
s_thread_para[i][0] = 0;//設置線程占用標志為"空閑"
s_thread_para[i][7] = i;//線程池索引
pthread_mutex_lock(s_mutex + i);//線程鎖
}

//創建線程池
for (i = 0; i < THREAD_MAX; i++)
{
rc = pthread_create(s_tid + i, 0, (void *)test_server4, (void *)(s_thread_para[i]));
if (0 != rc)
{
fprintf(stderr, "線程創建失敗\n");
return(-1);
}
}

//成功返回
return(0);
}

static int init_listen4(char *ip4, int port, int max_link)
{
//臨時變量
int sock_listen4;
struct sockaddr_in addr4;
unsigned int optval;
struct linger optval1;

//初始化數據結構
bzero(&addr4, sizeof(addr4));
inet_pton(AF_INET, ip4, &(addr4.sin_addr));
addr4.sin_family = AF_INET;
addr4.sin_port = htons(port);

//創建SOCKET
sock_listen4 = socket(AF_INET, SOCK_STREAM, 0);
if (0 > sock_listen4) return(-1);

//設置SO_REUSEADDR選項(服務器快速重起)
optval = 0x1;
setsockopt(sock_listen4, SOL_SOCKET, SO_REUSEADDR, &optval, 4);

//設置SO_LINGER選項(防范CLOSE_WAIT掛住所有套接字)
optval1.l_onoff = 1;
optval1.l_linger = 60;
setsockopt(sock_listen4, SOL_SOCKET, SO_LINGER, &optval1, sizeof(struct linger));

if (0 > bind(sock_listen4, (struct sockaddr *)&addr4, sizeof(addr4)))
{
close(sock_listen4);
return(-1);
}

if (0 > listen(sock_listen4, max_link))
{
close(sock_listen4);
return(-1);
}

return(sock_listen4);
}

void * test_server4(unsigned int thread_para[])
{
//臨時變量
int pool_index; //線程池索引
int sock_cli; //客戶端連接
int listen_index; //監聽索引

char buff[32768]; //傳輸緩沖區
char *p;
int i, j, len;

//線程脫離創建者
pthread_detach(pthread_self());
pool_index = thread_para[7];

wait_unlock:

pthread_mutex_lock(s_mutex + pool_index);//等待線程解鎖

//線程變量內容復制
sock_cli = thread_para[1];//客戶端連接
listen_index = thread_para[2];//監聽索引

//接收請求
len = recv(sock_cli, buff, 32768, MSG_NOSIGNAL);

//構造響應
p = buff;
//HTTP頭
p += sprintf(p, "HTTP/1.1 200 OK\r\n");
p += sprintf(p, "Content-Type: text/html\r\n");
p += sprintf(p, "Connection: closed\r\n\r\n");
//頁面
p += sprintf(p, "<html>\r\n<head>\r\n");
p += sprintf(p, "<meta content=\"text/html; charset=GBK\" http-equiv=\"Content-Type\">\r\n");
p += sprintf(p, "</head>\r\n");
p += sprintf(p, "<body style=\"background-color: rgb(229, 229, 229);\">\r\n");

p += sprintf(p, "<center>\r\n");
p += sprintf(p, "<H3>連接狀態</H3>\r\n");
p += sprintf(p, "<p>服務器地址 %s:%d</p>\r\n", s_listens[listen_index].ip4, s_listens[listen_index].port);
j = 0;
for (i = 0; i < THREAD_MAX; i++)
{
if (0 != s_thread_para[i][0]) j++;
}
p += sprintf(p, "<H3>線程池狀態</H3>\r\n");
p += sprintf(p, "<p>線程池總數 %d 活動線程總數 %d</p>\r\n", THREAD_MAX, j);
p += sprintf(p, "</center></body></html>\r\n");
len = p – buff;

//發送響應
send(sock_cli, buff, len, MSG_NOSIGNAL);

//釋放連接
shutdown(sock_cli, SHUT_RDWR);
close(sock_cli);

//線程任務結束
thread_para[0] = 0;//設置線程占用標志為"空閑"
goto wait_unlock;

pthread_exit(NULL);
}
[/cpp]

 

參考資料：

http://www.coder4.com/archives/144