RT，使用消息隊列，信號量和命名管道實現的多人群聊系統。

本學期Linux、unix網絡編程的第三個作業。

先上實驗要求：

實驗三? 多進程服務器

【實驗目的】

1、熟練掌握進程的創建與終止方法；

2、熟練掌握進程間通信方法；

2、應用套接字函數完成多進程服務器，實現服務器與客戶端的信息交互。

【實驗學時】

??? 4學時

【實驗內容】

通過一個服務器實現最多5個客戶之間的信息群發。

服務器顯示客戶的登錄與退出；

客戶連接后首先發送客戶名稱，之后發送群聊信息；

客戶輸入bye代表退出，在線客戶能顯示其他客戶的登錄于退出。

?

任務分析：

?

實現提示：

1、服務器端：

服務器進程稱之為主進程，主進程創建一個轉發子進程和最多5個通信子進程。

主進程與轉發子進程之間：

??????? 信號量（初值5，主進程接受一個客戶連接后執行P操作判斷是否超過5，轉發子進程有一個客戶退出后執行V操作,并發消息隊列標識符）

??????? 命名管道SERVER（轉發子進程將可用的消息隊列標識符寫入管道，主進程從管道中讀取消息隊列標識符）

轉發子進程與通信子進程之間：

??????? 命名管道CLIENT（通信子進程向命名管道寫入客戶端發來的消息，轉發子進程從管道中讀取消息并發送給對應的客戶端）

??????? 消息隊列（轉發子進程將客戶發來的信息通過消息隊列發送給每個通信子進程）

（1）主進程：

從轉發子進程獲取一個可用的消息隊列標識符；

接收客戶連接請求，如果連接數超過最大連接數，向客戶發送退出標志，否則發送OK標志；

每接受一個連接，創建一個通信子進程并將連接socket、消息隊列標識符、客戶地址傳遞給通信子進程。

（2）通信子進程：

創建一個子進程負責從消息隊列中讀取消息，發送給客戶。

通信子進程負責接收客戶發來信息，通過命名管道CLIENT發送給轉發子進程；

若信息為用戶名，附帶消息隊列、客戶地址發送給轉發子進程；

若信息為退出，終止子進程，程序結束

（3）轉發子進程：

創建5個消息隊列；

維護客戶信息表：消息隊列、客戶名、客戶IP、客戶端口、狀態。

從命名管道CLIENT中讀取通信子進程發來的消息，消息類型為：用戶名、退出及一般信息；

若為用戶名，依據消息隊列在更新客戶信息表，狀態為可用；

若為一般信息，將信息轉換后寫入可用客戶的消息隊列，等待其他通信子進程讀取；

若為退出，在客戶信息表中狀態設為不可用，執行信號量V操作，并將可用客戶的消息隊列標識符寫入到命名管道SERVER；

2、客戶端：

根據用戶從終端輸入的服務器IP地址及端口號連接到相應的服務器；

連接成功后，先發送客戶名稱；

創建一個子進程負責接收服務器發來的信息，并顯示；

主進程循環從終端輸入信息，并將信息發送給服務器；

當發送給服務器為bye后，關閉子進程，程序退出。

?

架構看起來很復雜，我們可以繪制一下流程圖方便理清思路。

?

在word里面截圖不是很清晰啊。。。

開始寫代碼吧：首先clientmsg.h，它定義了一些消息的操作符（OP）和CLIENTMSG這個結構體（用于服務器和客戶端之間傳遞消息）

1 //CLIENTMSG between server and client 2 #ifndef _clientmsg 3 #define _clientmsg 4 5 //USER MSG EXIT for OP of CLIENTMSG 6 #define EXIT -1 7 #define USER 1 8 #define MSG 2 9 #define OK 3 10 11 #ifndef CMSGLEN 12 #define CMSGLEN 100 13 #endif 14 15 struct CLIENTMSG{ 16 int OP; 17 char username[20]; 18 char buf[CMSGLEN]; 19 }; 20 21 #endif

然后實現一下servermsg.h，用于服務器內部的轉發子進程和通信子進程之間的消息傳遞。

1 //SERVERMSG for communicate to translate 2 //MESSAGE for translate to communicate 3 #ifndef _servermsg 4 #define _servermsg 5 6 #include <netinet/in.h> 7 #include "clientmsg.h" 8 9 10 #ifndef CMSGLEN 11 #define CMSGLEN 100 12 #endif 13 14 15 struct SERVERMSG{ 16 int OP; 17 char username[20]; 18 char buf[CMSGLEN]; 19 struct sockaddr_in client; 20 int stat; 21 int qid; 22 }; 23 24 struct MESSAGE{ 25 long msgtype; 26 struct SERVERMSG msg; 27 }; 28 29 #endif

由于需要操作信號量，所以將一些信號量的操作做成函數

semaphore.h

1 #ifndef _semaphore 2 #define _semaphore 3 4 union semun 5 { 6 int val; 7 struct semid_ds *buf; 8 unsigned short *array; 9 }; 10 11 int CreateSem(key_t key,int value); 12 int Sem_P(int semid); 13 int Sem_V(int semid); 14 int GetvalueSem(int semid); 15 void DestroySem(int semid); 16 17 18 #endif

對函數的實現：semaphore.c

1 #include <stdlib.h> 2 #include <fcntl.h> 3 #include <sys/sem.h> 4 #include "semaphore.h" 5 6 int CreateSem(key_t key,int value) 7 { 8 union semun sem; 9 int semid; 10 sem.val=value; 11 semid=semget(key,1,IPC_CREAT); 12 if (semid==-1){ 13 perror("semget error"); exit(1); 14 } 15 semctl(semid,0,SETVAL,sem); 16 return semid; 17 } 18 19 int Sem_P(int semid) 20 { 21 struct sembuf sops={0,-1,IPC_NOWAIT}; 22 return (semop(semid,&sops,1)); 23 } 24 25 int Sem_V(int semid) 26 { 27 struct sembuf sops={0,+1,IPC_NOWAIT}; 28 return (semop(semid,&sops,1)); 29 } 30 31 int GetvalueSem(int semid) 32 { 33 union semun sem; 34 return semctl(semid,0,GETVAL,sem); 35 } 36 void DestroySem(int semid) 37 { 38 union semun sem; 39 sem.val=0; 40 41 semctl(semid,0,IPC_RMID,sem); 42 }

?

接下來是非常重要的服務器端實現（里面有很多調試信息，比較懶沒有刪掉，直接在里面注釋掉了。）

server.c

1 #include <stdio.h> 2 #include <string.h> 3 #include <sys/socket.h> 4 #include <netinet/in.h> 5 #include <stdlib.h> 6 #include <sys/types.h> 7 #include <sys/wait.h> 8 #include <sys/stat.h> 9 #include <unistd.h> 10 #include <fcntl.h> 11 #include <sys/ipc.h> 12 #include "semaphore.h" 13 #include "servermsg.h" 14 15 16 void trans_process(int semid); 17 void communicate_process(int connetfd,int qid,struct sockaddr_in client); 18 19 int main(){ 20 21 struct sockaddr_in server; 22 struct sockaddr_in client; 23 int listenfd,connetfd; 24 char ip[20]; 25 int port; 26 int addrlen; 27 struct CLIENTMSG clientMsg; 28 int ret,status; 29 /*---------------------socket-------------------*/ 30 if((listenfd = socket(AF_INET,SOCK_STREAM,0))== -1){ 31 perror("socket() error\n"); 32 exit(1); 33 } 34 35 /*----------------------IO-----------------------*/ 36 printf("Please input the ip:\n"); 37 scanf("%s",ip); 38 printf("Please input the port:\n"); 39 scanf("%d",&port); 40 41 /*---------------------bind----------------------*/ 42 bzero(&server,sizeof(server)); 43 server.sin_family = AF_INET; 44 server.sin_port = htons(port); 45 server.sin_addr.s_addr = inet_addr(ip); 46 if(bind(listenfd,(struct sockaddr *)&server,sizeof(server))== -1){ 47 perror("bind() error\n"); 48 exit(1); 49 } 50 51 /*----------------------listen-------------------*/ 52 if (listen(listenfd,5)== -1){ 53 perror("listen() error\n"); 54 exit(1); 55 } 56 57 //創建命名管道 58 unlink("SERVER"); 59 mkfifo("SERVER",O_CREAT); 60 int rd = open("SERVER",O_RDONLY|O_NONBLOCK); 61 int semid; 62 key_t k = ftok(".",'b'); 63 semid = CreateSem(k,5); 64 pid_t pid_1,pid_2; 65 pid_1 = fork(); 66 if(pid_1 == 0){ 67 trans_process(semid); 68 exit(0); 69 } 70 else if(pid_1 > 0){ 71 while(1){ 72 addrlen = sizeof(client); 73 if((connetfd = accept(listenfd,(struct sockaddr *)&client,&addrlen))== -1){ 74 perror("accept() error\n"); 75 exit(1); 76 } 77 ret = Sem_P(semid); 78 if(ret == 0){ 79 int qid; 80 read(rd,&qid,sizeof(qid)); 81 //printf("qid1:%d\n",qid ); 82 pid_2 = fork(); 83 if (pid_2 > 0){ 84 close(connetfd); 85 waitpid(pid_2,&status,WNOHANG); 86 continue; 87 } 88 else if(pid_2 == 0){ 89 communicate_process(connetfd,qid,client); 90 exit(0); 91 } 92 else { 93 perror("the second fork error\n"); 94 } 95 } 96 else { 97 clientMsg.OP = EXIT; 98 send(connetfd,&clientMsg,sizeof(clientMsg),0); 99 close(connetfd); 100 } 101 waitpid(pid_1,&status,WNOHANG); 102 103 } 104 } 105 else { 106 perror("first time fork error\n"); 107 } 108 /*----------------------close-------------------*/ 109 close(connetfd); 110 close(listenfd); 111 112 return 0; 113 } 114 115 116 /*----------------------------函數實現區----------------------------*/ 117 void trans_process(int semid){ 118 struct SERVERMSG ent[5]; 119 struct MESSAGE sendMsg; 120 struct SERVERMSG msg; 121 int i; 122 for(i=0;i<5;i++){ 123 ent[i].stat = 0; 124 } 125 int wfd = open("SERVER",O_WRONLY|O_NONBLOCK); 126 for(i=0;i<5;i++){ 127 key_t key = ftok(".",(char)i+102); 128 ent[i].qid = msgget(key,IPC_CREAT); 129 write(wfd,&ent[i].qid,sizeof(ent[i].qid)); 130 } 131 unlink("CLIENT"); 132 mkfifo("CLIENT",O_CREAT); 133 int rfd = open("CLIENT",O_RDONLY|O_NONBLOCK); 134 int len; 135 while(1){ 136 bzero(&msg,sizeof(msg)); 137 len = read(rfd,&msg,sizeof(msg)); 138 //printf(" %d,%s ,%s\n",msg.OP,msg.username,msg.buf ); 139 //sleep(3); 140 if(len > 0){ 141 if(msg.OP == USER){ 142 for(i=0;i<5;i++){ 143 if(ent[i].qid == msg.qid){ 144 bcopy(msg.username,ent[i].username,strlen(msg.username)); 145 ent[i].client = msg.client; 146 ent[i].stat = 1; 147 break; 148 } 149 } 150 } 151 else if(msg.OP == EXIT){ 152 for(i=0;i<5;i++){ 153 if(ent[i].qid == msg.qid){ 154 ent[i].stat = 0; 155 write(wfd,&ent[i].qid,sizeof(ent[i].qid)); 156 Sem_V(semid); 157 break; 158 } 159 } 160 } 161 //bzero(&sendMsg,sizeof(sendMsg)); 162 sendMsg.msg = msg; 163 for(i=0;i<5;i++){ 164 if(ent[i].stat == 1){ 165 printf("stat 1...\n"); 166 int m_len = sizeof(msg); 167 int sta=msgsnd(ent[i].qid,&sendMsg,len,0); 168 //printf("flag:%d\n",sta ); 169 } 170 } 171 } 172 else { 173 continue; 174 } 175 } 176 177 } 178 179 void communicate_process(int connetfd,int qid,struct sockaddr_in client){ 180 struct CLIENTMSG sendMsg; 181 struct CLIENTMSG recvMsg; 182 struct MESSAGE server_Msg; 183 struct SERVERMSG client_sndMsg; 184 struct SERVERMSG msg; 185 int status; 186 int wfd = open("CLIENT",O_WRONLY|O_NONBLOCK); 187 pid_t pid; 188 pid = fork(); 189 if(pid < 0){ 190 perror("communicate_process fork error\n"); 191 } 192 else if (pid == 0){ 193 bzero(&sendMsg,sizeof(sendMsg)); 194 sendMsg.OP = OK; 195 send(connetfd,&sendMsg,sizeof(sendMsg),0); 196 while(1){ 197 int m_len = sizeof(msg); 198 bzero(&server_Msg,sizeof(server_Msg)); 199 int sta=msgrcv(qid,&server_Msg,m_len,0,0); 200 //printf("flag:%d\n",sta ); 201 //printf("send..%d,%s,%s\n",server_Msg.msg.OP,server_Msg.msg.username,server_Msg.msg.buf ); 202 bzero(&sendMsg,sizeof(sendMsg)); 203 bcopy(server_Msg.msg.username,sendMsg.username,strlen(server_Msg.msg.username)); 204 sendMsg.OP = server_Msg.msg.OP; 205 bcopy(server_Msg.msg.buf,sendMsg.buf,strlen(server_Msg.msg.buf)); 206 //printf("send..%d,%s,%s\n",sendMsg.OP,sendMsg.username,sendMsg.buf ); 207 send(connetfd,&sendMsg,sizeof(sendMsg),0); 208 } 209 } 210 else{ 211 while(1){ 212 bzero(&recvMsg,sizeof(recvMsg)); 213 int len =recv(connetfd,&recvMsg,sizeof(recvMsg),0); 214 if(len > 0){ 215 if(recvMsg.OP == USER){ 216 printf("user %s login from ip:%s,port:%d\n",recvMsg.username,inet_ntoa(client.sin_addr),ntohs(client.sin_port) ); 217 client_sndMsg.OP = USER; 218 } 219 else if(recvMsg.OP == EXIT){ 220 printf("user %s is logout\n",recvMsg.username ); 221 client_sndMsg.OP = EXIT; 222 write(wfd,&client_sndMsg,sizeof(client_sndMsg)); 223 break; 224 } 225 else if(recvMsg.OP == MSG){ 226 client_sndMsg.OP = MSG; 227 } 228 bzero(&client_sndMsg,sizeof(client_sndMsg)); 229 bcopy(recvMsg.username,client_sndMsg.username,strlen(recvMsg.username)); 230 bcopy(recvMsg.buf,client_sndMsg.buf,strlen(recvMsg.buf)); 231 client_sndMsg.client = client; 232 //printf("qid2:%d\n",qid ); 233 client_sndMsg.qid = qid; 234 client_sndMsg.OP = recvMsg.OP; 235 write(wfd,&client_sndMsg,sizeof(client_sndMsg)); 236 237 } 238 else{ 239 continue; 240 } 241 } 242 kill(pid,SIGKILL); 243 waitpid(pid,&status,WNOHANG); 244 close(wfd); 245 close(connetfd); 246 } 247 }

?

寫出了服務端，就可以非常容易的寫出客戶端了。

client.c

1 #include <stdio.h> 2 #include <string.h> 3 #include <sys/socket.h> 4 #include <netinet/in.h> 5 #include <stdlib.h> 6 #include <sys/types.h> 7 #include <sys/wait.h> 8 #include <signal.h> 9 #include <unistd.h> 10 #include "clientmsg.h" 11 12 int main(){ 13 int sockfd; 14 char ip[20]; 15 int port; 16 int status; 17 pid_t pid; 18 struct sockaddr_in server; 19 struct CLIENTMSG clientMsg; 20 21 /*---------------------socket---------------------*/ 22 if((sockfd = socket(AF_INET,SOCK_STREAM,0))== -1){ 23 perror("socket error\n"); 24 exit(1); 25 } 26 27 /*---------------------connect--------------------*/ 28 printf("Please input the ip:\n"); 29 scanf("%s",ip); 30 printf("Please input the port:\n"); 31 scanf("%d",&port); 32 bzero(&server,sizeof(server)); 33 server.sin_family = AF_INET; 34 server.sin_port = htons(port); 35 inet_aton(ip,&server.sin_addr); 36 if(connect(sockfd,(struct sockaddr *)&server,sizeof(server))== -1){ 37 perror("connect() error\n"); 38 exit(1); 39 } 40 recv(sockfd,&clientMsg,sizeof(clientMsg),0); 41 if(clientMsg.OP == OK){ 42 int len; 43 pid = fork(); 44 if(pid == 0){ 45 while(1){ 46 bzero(&clientMsg,sizeof(clientMsg)); 47 len =recv(sockfd,&clientMsg,sizeof(clientMsg),0); 48 if(len > 0){ 49 if(clientMsg.OP ==USER){ 50 printf("the user %s is login.\n",clientMsg.username ); 51 } 52 else if(clientMsg.OP == EXIT){ 53 printf("the user %s is logout.\n",clientMsg.username); 54 } 55 else if(clientMsg.OP == MSG){ 56 printf("%s: %s\n",clientMsg.username,clientMsg.buf ); 57 } 58 } 59 } 60 exit(EXIT_SUCCESS); 61 } 62 else if(pid > 0){ 63 printf("Please input the username:\n"); 64 scanf("%s",clientMsg.username); 65 clientMsg.OP = USER; 66 send(sockfd,&clientMsg,sizeof(clientMsg),0); 67 while(1){ 68 clientMsg.OP = MSG; 69 scanf("%s",clientMsg.buf); 70 if(strcmp("bye",clientMsg.buf) == 0){ 71 clientMsg.OP = EXIT; 72 send(sockfd,&clientMsg,sizeof(clientMsg),0); 73 break; 74 } 75 send(sockfd,&clientMsg,sizeof(clientMsg),0); 76 77 } 78 kill(pid,SIGKILL); 79 waitpid(pid,&status,WNOHANG); 80 } 81 else{ 82 perror("fork error!\n"); 83 } 84 } 85 else{ 86 printf("以達到最大連接數！\n"); 87 } 88 /*------------------------close--------------------------*/ 89 close(sockfd); 90 91 return 0; 92 }

最后是makefile：

main:server.o client.o semaphore.ogcc server.o semaphore.o -oservergcc client.o -oclient server.o:server.c semaphore.h clientmsg.h servermsg.hgcc -c server.c client.o:client.c clientmsg.hgcc -c client.c semaphore.o:semaphore.h semaphore.cgcc -c semaphore.c clean:rm -rf *.o

?

?

下面上一下演示過程：（測試環境，Red Hat Enterprise Linux 6 + centos系Linux，ubuntu下可能會有些問題。）

首先先把服務端啟動開來，為了方便測試，這里直接使用的是127.0.0.1的localhost。

?

然后啟動兩個客戶端用來測試，在用戶登錄的時候客戶端會有消息提醒。服務端會有日志打印輸出客戶端的名字和登錄ip、端口。

客戶可以發送消息了，如圖發送與接收均正常。可以同時啟動<=5個客戶端進行群聊，這里為了簡單演示只是啟動了2個。（修改信號量代碼可以實現n多個客戶的同時登陸）：

?

輸入bye以后即可退出聊天并下線。當有客戶下線的時候，在線的客戶端會收到下線提醒，客戶端會有日志打印輸出。

?

?

這個實驗內容前前后后花了我2天才寫完，剛開始沒有弄清楚這一整套的工作機制與流程，寫起來很是吃力，程序就是各種調不通。本來都想放棄了，但是后來還是咬咬牙堅持了一下來，飯要一口一口吃，程序要一點一點的寫，萬事不能操之過急，寫代碼一定要心平氣和，頭腦清晰。由于gdb調試工具用的不是很熟練，只能在程序里面一段一段的print變量來DEBUG，很是辛苦啊。

轉載于:https://www.cnblogs.com/msxh/p/4943755.html

總結

以上是生活随笔為你收集整理的【LINUX/UNIX网络编程】之使用消息队列，信号量和命名管道实现的多进程服务器（多人群聊系统）...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。