- 一、实验目的
- 二、实验内容
- 三、实验步骤
- 1.设计服务器和客户端的编程思路：
- 2.设计程序测试步骤与预期结果：
- 四、程序以及运行结果
- 1.程序部分
- 2.步骤-运行结果
- 总结
一、实验目的
在Linux系统中，使用C语言编程设计一个基于TCP/IP的多线程echo服务器程序，从而实现对客户端发送内容的响应。主要实现为一个面向连接的、多线程并发服务器，并由客户端程序发送信息。
二、实验内容修改多线程的echo服务器程序，其中客户端能将相应信息传输给服务器，服务器从客户端接收信息，并用将请求的信息内容应答客户端。同时要求一个服务器能够为多个客户并发提供服务，使之能够满足：
- 服务器用多线程来实现并发；
- 在程序开始时创建固定数目的线程；
- 当客户连接请求到来时，不会创建服务于客户端的线程，而是创建连接并分配一个事先创建的线程为客户提供服务；
- 如果连接数多于预先创建的线程数，则新创建的连接客户进行排队，等待最近的连接被释放后，再利于此空闲线程进行连接。
- 服务器通过套接字编程，已经设计好如何进行TCP传输响应信息，将echo服务器程序的几个文件内容合并起来，通过线程池的设计满足以上所有的实现需求。
- 首先解释线程池的设计内容，其包括线程执行队列（struct WORKER）、任务队列（struct JOB）、管理者组件（struct MANAGER）、条件变量（pthread_cond_t）、锁（pthread_mutex_t）和任务队列计数（num）。
- 含有固定数目线程的线程池需要在一个主函数中被创建，通过设计其条件变量、互斥锁等内容对其进行初始化，循环调用线程创建函数从而创建对应数量的线程，主线程将已经创建的线程以链表的形式连接成队列，同时子线程进入相应线程函数。
- 在子线程函数中，先进行加锁等待任务队列不为空时条件等待的同步，当有连接相应，即在服务器main函数中创建了新的任务加入到任务队列时，唤醒等待，并从线程队列中选取（若存在空闲线程的话）线程执行echo功能。若没有空间线程，则继续等待到其他线程释放。
- 所以此处，线程池起到一个管理整个预制数量线程的一个作用，符合题目要求和实现。
- 客户端从终端接收IP地址和端口号；然后创建流式套接字；将套接字设置为IPv4、从终端读入的端口号以及IP地址；通过同一个套接字进行服务器连接；客户端若从终端接收到退出请求，则退出并套接字；客户端若每从终端接受一个信息，就通过向服务器发送一条信息，然后将等待服务器端的响应信息；收到响应信息后进入下一轮等待。
四、程序以及运行结果 1.程序部分同下四-2
服务器：
//多线程服务器代码 //createby:jty //TCPmetechod.c的头文件 #include#include #include #include #include #include #include #include #include #include #include #include #include //errexit.c的头文件 #include //passivesock.c的头文件 #include #include // #define WORKERS 2 //TCPmetechod.c的宏定义 #define QLEN 32 #define BUFSIZE 4096 #define INTERVAL 5 //线程池函数宏定义： //尾插法添加一个结点入队列 #define LL_ADD(item, list) do { if (list == NULL){ list=item; } else{ item->prev = list; while(list->next != NULL){ list = list->next; } list->next = item; item->next = item->prev; item->prev = list; list = item->next; item->next = NULL; } } while(0) //将一个结点从队列中删除 #define LL_REMOVE(item, list) do { if (item->prev != NULL) item->prev->next = item->next; if (item->next != NULL) item->next->prev = item->prev; if (item == list) list = item->next; item->prev = item->next = NULL; } while(0) extern int errno; unsigned short portbase = 0; //状态输出结构体 struct { pthread_mutex_t st_mutex; unsigned int st_concount; unsigned int st_contotal; unsigned long st_contime; unsigned long st_bytecount; } stats; typedef struct MANAGER ThreadPool; typedef struct JOB JOB; //执行队列结构体 struct WORKER { pthread_t thread; ThreadPool *pool; int terminate; //停止工作 struct WORKER *next; struct WORKER *prev; }; //任务队列结构体 struct JOB { void * (*func)(void *arg); //任务函数 void *user_data; //任务函数的参数 struct JOB *next; struct JOB *prev; }; //管理组件结构体 struct MANAGER { struct WORKER *workers; //执行队列 struct JOB *jobs; //任务队列 int num; pthread_cond_t jobs_cond; //任务队列条件等待变量 pthread_mutex_t jobs_mutex; //任务队列加锁 }; //线程池引入函数 int threadPoolCreate(ThreadPool *pool, int numWorkers); int threadPoolDestory(ThreadPool *pool); void threadPoolPush(ThreadPool *pool, void *(*func)(void *arg), void *arg); //TCPmtechod.c中函数 void prstats(void); int TCPechod(int fd); int errexit(const char *format, ...); int passiveTCP(const char *service, int qlen); //passiveTCP.c中函数 int passivesock(const char *service, const char *transport, int qlen); int main(int argc, char *argv[]) { char *service = "echo"; struct sockaddr_in fsin; unsigned int alen; int msock; int ssock; //判断输入合法 switch (argc) { case 1: break; case 2: service = argv[1]; break; default: errexit("usage: TCPechod [port]n"); } msock = passiveTCP(service, QLEN); //创建线程池 ThreadPool pool; int workers = WORKERS; int ret = threadPoolCreate(&pool,workers); if(ret < 0){ fprintf(stdout, "threadpool_create failed!n"); return ret; } while (1) { alen = sizeof(fsin); ssock = accept(msock, (struct sockaddr *)&fsin, &alen); if(ssock < 0){ if(errno == EINTR) continue; errexit("accept: %sn", strerror(errno)); } threadPoolPush(&pool,TCPechod, ssock); } threadPoolDestory(&pool); } static void* threadCallback(void *args) { struct WORKER *worker = (struct WORKER*)args; //worker有两种状态：执行或是等待 while (1) { //条件等待加锁 pthread_mutex_lock(&worker->pool->jobs_mutex); //若任务队列为空 while(worker->pool->jobs == NULL){ //终止信号 if(worker->terminate) break; //条件等待 pthread_cond_wait(&worker->pool->jobs_cond, &worker->pool->jobs_mutex); } //终止信号退出 if(worker->terminate){ //条件等待解锁 pthread_mutex_unlock(&worker->pool->jobs_mutex); break; } //打印提示信息 printf("n~~~~~~~~~~~~~~~~Thread %lu is awake!~~~~~~~~~~~~~~~~nn",worker->thread); struct JOB *job = worker->pool->jobs; LL_REMOVE(job, worker->pool->jobs); //条件等待解锁 pthread_mutex_unlock(&worker->pool->jobs_mutex); //执行响应函数 job->func(job->user_data); worker->pool->num--; } free(worker); pthread_exit(NULL); //线程退出 } int threadPoolCreate(ThreadPool *pool, int numWorkers) { pool->num = 0; pthread_t th; pthread_attr_t ta; //线程属性 (void) pthread_attr_init(&ta); (void) pthread_attr_setdetachstate(&ta, PTHREAD_CREATE_DETACHED); (void) pthread_mutex_init(&stats.st_mutex, 0); //输出显示线程创建 if (pthread_create(&th, &ta, (void * (*)(void *))prstats, 0) < 0) errexit("pthread_create(prstats): %sn", strerror(errno)); if(numWorkers < 1) numWorkers = 1; if(pool == NULL) return -1; memset(pool, 0, sizeof(ThreadPool)); pool->workers=NULL; pool->jobs=NULL; //初始化条件变量和互斥锁 //结构体赋值--内存拷贝 pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER; memcpy(&pool->jobs_cond, &blank_cond, sizeof(pthread_cond_t)); pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER; memcpy(&pool->jobs_mutex, &blank_mutex, sizeof(pthread_mutex_t)); fprintf(stdout, "n--------------------------------------------------------------------------------n"); //初始化执行队列 int i = 0; for (i = 0; i < numWorkers; i++) { struct WORKER *worker = (struct WORKER *)malloc(sizeof(struct WORKER)); if (worker == NULL) { perror("malloc"); return -2; } memset(worker, 0, sizeof(struct WORKER)); worker->next=NULL; worker->prev=NULL; worker->pool = pool; //线程指针变量，属性设置为空，线程执行函数，传入执行函数的参数 int ret = pthread_create(&worker->thread, &ta, threadCallback, worker); if (ret) { perror("pthread_create"); free(worker); return -3; } fprintf(stdout, "worker thread: %lun", worker->thread); //形成一个执行队列 LL_ADD(worker, pool->workers); } fprintf(stdout, "--------------------------------------------------------------------------------n"); return 0; } void threadPoolPush(ThreadPool *pool, void *(*func)(void *arg), void *arg) { //创建一个任务 JOB *job=(JOB*)malloc(sizeof(JOB)); if(job==NULL){ perror("malloc"); exit(1); } memset(job, 0, sizeof(JOB)); job->func = func; job->user_data = arg; job->next=NULL; job->prev=NULL; //将任务加入到线程池中 pthread_mutex_lock(&pool->jobs_mutex); if(job!=NULL) LL_ADD(job, pool->jobs); pool->num++; if(pool->num > WORKERS){ printf("n~~~~~~~~~~~~~~~~Waiting client number is %d.~~~~~~~~~~~~~~~~nn",(pool->num - WORKERS)); fflush(stdout); } else{ pthread_cond_signal(&pool->jobs_cond); //唤醒一个线程 } pthread_mutex_unlock(&pool->jobs_mutex); } int threadPoolDestory(ThreadPool *pool) { struct WORKER *worker = NULL; for (worker = pool->workers; worker != NULL; worker = worker->next){ worker->terminate = 1; } //唤醒所有线程 pthread_mutex_lock(&pool->jobs_mutex); int ret = pthread_cond_broadcast(&pool->jobs_cond); pthread_mutex_unlock(&pool->jobs_mutex); return ret; } int TCPechod(int fd) { time_t start; char buf[BUFSIZ]; int cc; start = time(0); (void) pthread_mutex_lock(&stats.st_mutex); stats.st_concount++; (void) pthread_mutex_unlock(&stats.st_mutex); memset(buf,0,sizeof(buf)); while (cc = read(fd, buf, sizeof buf)) { if (cc < 0) errexit("echo read: %sn", strerror(errno)); if (write(fd, buf, cc) <= 0) errexit("echo write: %sn", strerror(errno)); (void) pthread_mutex_lock(&stats.st_mutex); stats.st_bytecount += cc; (void) pthread_mutex_unlock(&stats.st_mutex); memset(buf,0,sizeof(buf)); } (void) close(fd); (void) pthread_mutex_lock(&stats.st_mutex); stats.st_contime += time(0) - start; stats.st_concount--; stats.st_contotal++; (void) pthread_mutex_unlock(&stats.st_mutex); return 0; } void prstats(void) { time_t now; while (1) { (void) sleep(INTERVAL); (void) pthread_mutex_lock(&stats.st_mutex); now = time(0); (void) printf("--- %s", ctime(&now)); (void) printf("%-32s: %un", "Current connections", stats.st_concount); (void) printf("%-32s: %un", "Completed connections", stats.st_contotal); if (stats.st_contotal) { (void) printf("%-32s: %.2f (secs)n", "Average complete connection time", (float)stats.st_contime / (float)stats.st_contotal); (void) printf("%-32s: %.2fn", "Average byte count", (float)stats.st_bytecount / (float)(stats.st_contotal + stats.st_concount)); } (void) printf("%-32s: %lunn", "Total byte count", stats.st_bytecount); (void) pthread_mutex_unlock(&stats.st_mutex); } } int errexit(const char *format, ...) { va_list args; va_start(args, format); vfprintf(stderr, format, args); va_end(args); exit(1); } int passivesock(const char *service, const char *transport, int qlen) { struct servent *pse; struct protoent *ppe; struct sockaddr_in sin; int s, type; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; sin.sin_addr.s_addr = INADDR_ANY; if ( pse = getservbyname(service, transport) ) sin.sin_port = htons(ntohs((unsigned short)pse->s_port) + portbase); else if ((sin.sin_port=htons((unsigned short)atoi(service))) == 0) errexit("can't get "%s" service entryn", service); if ( (ppe = getprotobyname(transport)) == 0) errexit("can't get "%s" protocol entryn", transport); if (strcmp(transport, "udp") == 0) type = SOCK_DGRAM; else type = SOCK_STREAM; s = socket(PF_INET, type, ppe->p_proto); if (s < 0) errexit("can't create socket: %sn", strerror(errno)); if (bind(s, (struct sockaddr *)&sin, sizeof(sin)) < 0) errexit("can't bind to %s port: %sn", service, strerror(errno)); if (type == SOCK_STREAM && listen(s, qlen) < 0) errexit("can't listen on %s port: %sn", service, strerror(errno)); return s; } int passiveTCP(const char *service, int qlen) { return passivesock(service, "tcp", qlen); }
客户端：
#include2.步骤-运行结果#include #include #include #include #include #include #include #include int main(int argc,char *argv[]) { int sockfd; int confd; struct hostent *he; //找到IP地址 struct sockaddr_in addr_ser; //服务器的地址 struct sockaddr_in addr_file; //服务器发来的文件地址 char sendMsg[1024],recvMsg[1024]; if(argc!=3){ //如果没有写服务器的连接 perror("Please input the server port!n"); exit(1); } he=gethostbyname((char *)argv[1]); //通过域名获取IP地址 if(he==NULL){ //如果获取IP地址失败 perror("Cannot get host by name!n"); exit(1); } sockfd=socket(AF_INET,SOCK_STREAM,0); //创建socket套接字 if(sockfd==-1){ //如果创建套接字失败 perror("Create socketfd failed!n"); exit(1); } memset(&addr_ser,0,sizeof(addr_ser)); addr_ser.sin_family = AF_INET; addr_ser.sin_port = htons((unsigned short)atoi(argv[2])); addr_ser.sin_addr = *((struct in_addr *) he->h_addr); confd = connect(sockfd, (struct sockaddr *)&addr_ser, sizeof(addr_ser)); //客户端连接服务器 if(confd == -1){ //如果连接失败 perror("Connectfd error!n"); exit(1); } while(1){ //循环使一个客户端不停的发送消息直到退出,同时建立新的连接 printf("--------------------------------------------------n"); memset(sendMsg,0,sizeof(sendMsg)); printf("Please input what you want to send: "); scanf("%s",sendMsg); if(strncmp(sendMsg,"EXIT",4)==0){ break; } int n1; n1=write(sockfd,sendMsg,strlen(sendMsg)); if(n1==0){ perror("client send wrong:"); break; } memset(recvMsg,0,sizeof(recvMsg)); int n2; n2=read(sockfd,recvMsg,1024); if(n2==-1){ perror("client recv wrong:"); break; } printf("Receive from server:%sn",recvMsg); fflush(stdout); } close(sockfd); return 0; }
此实验中，设计最大的线程数量为2，开启一个服务器和四个客户端进行测试。
编译服务器server，打开服务器，输入端口号——输出创建的两个线程的线程号
等待5秒钟，prstats进程打印相关的服务器连接和echo信息的数据——由于没有客户端连接，所以结果为0，0，0
打开一个客户端，输入要连接的IP地址和端口号——服务器端显示分配给该响应的线程号
输入响应信息——服务器端输出已连接的客户端个数，客户端收到服务器的回声信息
打开第二个客户端，输入要连接的IP地址和端口号——服务器端显示分配给该响应的线程号
在第二个客户端输入响应信息，并启动第三个客户端——客户端2收到响应信息，服务器端显示两个连接和等待连接的客户端个数1
在第三个客户端中输入信息——服务器无响应
启动第四个客户端——服务器端显示两个连接和等待连接的客户端个数2
在第四个客户端中输入信息——服务器无响应
退出第一个客户端——第三个客户端马上收到回应，服务器端显示被分配的线程号，并展示有2个客户端正在连接，1个已经结束连接，平均连接时长、传输比特数等等。
在第三个客户端中测试输入信息——可以成功从服务器端收获回应信息
退出第二个客户端——第四个客户端马上收到回应，服务器端显示被分配的线程号，并展示有2个客户端正在连接，2个已经结束连接，平均连接时长、传输比特数等等。
在第四个客户端中测试输入信息——可以成功从服务器端收获回应信息
总结
喜欢这篇文章的话【点赞】、【收藏】下呗！
