sgzwiz
- 浏览: 2227 次
-
文章分类
最新评论
epoll服务器示例, 监听5000个端口, 使用线程池
运行这个程序需要预先设置栈内存和文件描述符上限, 否则运行失败
ulimit -n 16384
ulimit -s 4096
文件名:server.c
编译: gcc server.c -Wall -O2 -pthread -o server
程序源码如下(请自行编辑宏定义SERVER_IP为自己的IP):
/*Linux 2.6 x86_64 only*/
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.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>
#define THREAD_MAX 4096
#define LISTEN_MAX 5000
#define SERVER_IP "192.168.1.103"
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 = 8000 + i;
//创建监听
rc = init_listen4(s_listens[i].ip4, s_listens[i].port, 64);
if (0 > rc) {
fprintf(stderr, "无法创建服务器监听于%s:%d\r\n", s_listens[i].ip4, s_listens[i].port);
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=UTF-8\" 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);
}
#include <iostream>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <pthread.h>
#define MAXLINE 10
#define OPEN_MAX 100
#define LISTENQ 20
#define SERV_PORT 5555
#define INFTIM 1000
//线程池任务队列结构体
struct task{
int fd; //需要读写的文件描述符
struct task *next; //下一个任务
};
//用于读写两个的两个方面传递参数
struct user_data{
int fd;
unsigned int n_size;
char line[MAXLINE];
};
//线程的任务函数
void * readtask(void *args);
void * writetask(void *args);
//声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件
struct epoll_event ev,events[20];
int epfd;
pthread_mutex_t mutex;
pthread_cond_t cond1;
struct task *readhead=NULL,*readtail=NULL,*writehead=NULL;
void setnonblocking(int sock)
{
int opts;
opts=fcntl(sock,F_GETFL);
if(opts<0)
{
perror("fcntl(sock,GETFL)");
exit(1);
}
opts = opts|O_NONBLOCK;
if(fcntl(sock,F_SETFL,opts)<0)
{
perror("fcntl(sock,SETFL,opts)");
exit(1);
}
}
int main()
{
int i, maxi, listenfd, connfd, sockfd,nfds;
pthread_t tid1,tid2;
struct task *new_task=NULL;
struct user_data *rdata=NULL;
socklen_t clilen;
pthread_mutex_init(&mutex,NULL);
pthread_cond_init(&cond1,NULL);
//初始化用于读线程池的线程
pthread_create(&tid1,NULL,readtask,NULL);
pthread_create(&tid2,NULL,readtask,NULL);
//生成用于处理accept的epoll专用的文件描述符
epfd=epoll_create(256);
struct sockaddr_in clientaddr;
struct sockaddr_in serveraddr;
listenfd = socket(AF_INET, SOCK_STREAM, 0);
//把socket设置为非阻塞方式
setnonblocking(listenfd);
//设置与要处理的事件相关的文件描述符
ev.data.fd=listenfd;
//设置要处理的事件类型
ev.events=EPOLLIN|EPOLLET;
//注册epoll事件
epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
bzero(&serveraddr, sizeof(serveraddr));
serveraddr.sin_family = AF_INET;
char *local_addr="200.200.200.222";
inet_aton(local_addr,&(serveraddr.sin_addr));//htons(SERV_PORT);
serveraddr.sin_port=htons(SERV_PORT);
bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));
listen(listenfd, LISTENQ);
maxi = 0;
for ( ; ; ) {
//等待epoll事件的发生
nfds=epoll_wait(epfd,events,20,500);
//处理所发生的所有事件
for(i=0;i<nfds;++i)
{
if(events[i].data.fd==listenfd)
{
connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);
if(connfd<0){
perror("connfd<0");
exit(1);
}
setnonblocking(connfd);
char *str = inet_ntoa(clientaddr.sin_addr);
std::cout<<"connec_ from >>"<<str<<std::endl;
//设置用于读操作的文件描述符
ev.data.fd=connfd;
//设置用于注测的读操作事件
ev.events=EPOLLIN|EPOLLET;
//注册ev
epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
}
else if(events[i].events&EPOLLIN)
{
printf("reading!\n");
if ( (sockfd = events[i].data.fd) < 0) continue;
new_task=new task();
new_task->fd=sockfd;
new_task->next=NULL;
//添加新的读任务
pthread_mutex_lock(&mutex);
if(readhead==NULL)
{
readhead=new_task;
readtail=new_task;
}
else
{
readtail->next=new_task;
readtail=new_task;
}
//唤醒所有等待cond1条件的线程
pthread_cond_broadcast(&cond1);
pthread_mutex_unlock(&mutex);
}
else if(events[i].events&EPOLLOUT)
{
rdata=(struct user_data *)events[i].data.ptr;
sockfd = rdata->fd;
write(sockfd, rdata->line, rdata->n_size);
delete rdata;
//设置用于读操作的文件描述符
ev.data.fd=sockfd;
//设置用于注测的读操作事件
ev.events=EPOLLIN|EPOLLET;
//修改sockfd上要处理的事件为EPOLIN
epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
}
}
}
}
void * readtask(void *args)
{
int fd=-1;
unsigned int n;
//用于把读出来的数据传递出去
struct user_data *data = NULL;
while(1){
pthread_mutex_lock(&mutex);
//等待到任务队列不为空
while(readhead==NULL)
pthread_cond_wait(&cond1,&mutex);
fd=readhead->fd;
//从任务队列取出一个读任务
struct task *tmp=readhead;
readhead = readhead->next;
delete tmp;
pthread_mutex_unlock(&mutex);
data = new user_data();
data->fd=fd;
if ( (n = read(fd, data->line, MAXLINE)) < 0) {
if (errno == ECONNRESET) {
close(fd);
} else
std::cout<<"readline error"<<std::endl;
if(data!=NULL)delete data;
} else if (n == 0) {
close(fd);
printf("Client close connect!\n");
if(data!=NULL)delete data;
} else{
data->n_size=n;
//设置需要传递出去的数据
ev.data.ptr=data;
//设置用于注测的写操作事件
ev.events=EPOLLOUT|EPOLLET;
//修改sockfd上要处理的事件为EPOLLOUT
epoll_ctl(epfd,EPOLL_CTL_MOD,fd,&ev);
}
}
}
分享到：
发表评论
相关推荐
本示例提供了一个基于epoll和线程池实现的简单网络服务器框架。通过合理利用epoll和线程池的技术优势，程序能够在处理高并发连接时保持良好的性能。通过对代码的深入理解，开发者可以进一步优化和扩展此框架，以满足...
一个简短的epoll服务器示例, 监听5000个端口, 使用线程池 运行这个程序需要预先设置栈内存和文件描述符上限, 否则运行失败
总结来说，这个Linux下的epoll TCP服务器示例涵盖了网络编程的基础要素，通过epoll机制实现了高效的事件驱动模型，适合学习和理解I/O多路复用技术在实际应用中的使用。同时，`Client.cpp`文件的存在使我们可以自测...
"基于C语言编写的高并发Epoll服务器"是一个典型的示例，它展示了如何利用C语言的强大功能来构建能够处理大量并发连接的高效服务器。Epoll是Linux内核提供的一种I/O多路复用技术，它在处理大量并发连接时表现出了卓越...
这个压缩包可能包含了一个示例代码，演示了如何在Linux服务器端使用epoll处理TCP连接，并在Windows客户端进行相应的socket通信。通过分析和运行这个例子，可以深入理解epoll和TCP在实际应用中的工作原理和实现方式。
当新的连接到来时，服务器会为每个连接创建一个新的线程或者使用线程池处理。这里，多线程技术被用来并行处理多个客户端连接，以提高服务效率。每个线程都会有一个独立的epoll实例，这样可以避免因一个连接阻塞而...
文件"2-1-1-multi-io"很可能包含了具体使用这些技术的代码示例，例如创建服务器、监听端口、注册文件描述符、处理I/O事件等。通过分析和学习这些代码，开发者可以更好地理解和掌握网络I/O多路复用的实际应用。 总的...
服务器通常会监听一个套接字，当新的连接请求到达时，服务器会为每个连接创建一个新的套接字，并将其添加到Epoll实例中。之后，服务器通过`epoll_wait()`监控所有连接的活动，当某个连接有数据可读或可写时，服务器...
本示例代码展示了如何使用 `epoll` 来构建一个简单的服务器程序，该服务器能够高效地处理大量并发连接。通过将监听套接字设置为非阻塞模式，并利用 `epoll` 的事件通知机制，可以显著提高服务器的响应能力和吞吐量。...
本实例“epoll_client_server.zip”提供了一个生动的示例，展示了如何使用epoll这一I/O多路复用机制来实现客户端和服务器端的通信。下面将详细介绍epoll、socket、TCP/IP以及它们在该压缩包中的应用。 首先，我们要...
- **epoll**常用于高性能服务器，如Web服务器、数据库服务器等，如Nginx、Redis等使用了epoll。 - **iocp**在Windows平台的服务器软件中常见，例如IIS、SQL Server等。 文件列表中，`iocp.cc`、`epoll_td2.cc`、`...
- 使用`epoll`，服务器可以为每个新连接创建一个单独的线程或使用线程池，确保资源有效利用，避免了`select`和`poll`在文件描述符数量过多时的性能下降问题。 4. **与Windows_socket比较**： - Windows提供了类似...
在给定的`python_epoll_test.tar.gz`压缩包中，包含了一个使用Python实现的`epoll`服务器和客户端的交互示例。Python作为一门高级编程语言，拥有丰富的库支持网络编程，如`socket`和`selectors`模块，其中`selectors...
本示例着重讲解了使用C++和epoll机制实现的client-server socket通信。Epoll是Linux内核提供的一种高效I/O事件通知机制，尤其适用于高并发的网络服务。下面我们将详细探讨相关知识点。 1. **Socket基础**： Socket...
3. 事件驱动模型：使用`epoll`，在一个单独的线程或进程内处理所有连接。当有事件发生时，通过`epoll_wait`获取并处理。 4. `epoll`的TCP服务器模型：结合`epoll`的“非阻塞I/O”和“事件驱动”特性，实现高效并发...
`socket()`用于创建一个Socket描述符，`bind()`将Socket绑定到特定的IP地址和端口，`listen()`使Socket进入监听状态，准备接收连接请求，而`accept()`则用于接收客户端的连接请求。 然而，在处理大量并发连接时，...
3. **设置epoll**：调用`epoll_create()`创建一个`epoll`实例，然后使用`epoll_ctl()`将刚才创建的UDP套接字添加到`epoll`实例中，设置为监听读事件。 4. **接收数据**：在主循环中，调用`epoll_wait()`阻塞等待事件...
以下是一个简单的epoll实现的echo服务器示例： ```c #include <sys/epoll.h> #include #include #include #include #include int main() { int server_sockfd, client_sockfd; struct sockaddr_in server_...
此外，对于多线程或异步I/O的UDP服务器，还需要考虑如何处理并发连接和数据处理，这通常会涉及线程池或事件驱动模型如epoll。 总的来说，这个"udp通信示例程序"是一个很好的起点，可以帮助开发者理解UDP通信的基本...