System V 消息队列—复用消息

2022-05-18 07:53:26

　　消息队列中的消息结构可以由我们*定义，具备较强的灵活性。通过消息结构可以共享一个队列，进行消息复用。通常定义一个类似如下的消息结构：

#define MSGMAXDAT     1024
struct mymsg
{
    long msg_len;   //消息长度
    long msg_type; //消息类型
    long msg_data[MSGMAXDATA]; //消息内容
};

消息结构相关联的类型字段（msg_type）提供了两个特性：

（1）标识消息，使得多个进程在单个队列上复用消息。

（2）用作优先级字段，允许接收者以不同于先进先出的某个顺序读出各个消息。

例子1：每个应用一个队列，可以在多个客户和单个服务器之间复用消息。使用一个消息队列进行通信，由消息类型标识消息是从客户到服务器，还是服务器到客户。通信模型如下：

按照通信模型编写程序如下：

公共头文件svmsg.h

#ifndef  SVMSG_H
#define  SVMSG_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <errno.h>

#define MSG_R 0400 /* read permission */
#define MSG_W 0200 /* write permission */
#define SVMSG_MODE (MSG_R | MSG_W | MSG_R >>3 | MSG_R >>6)
#define MQ_KEY  1234L
#define MSGMAX  1024
//消息结构
struct mymesg
{
   long mesg_len;
   long mesg_type;
   char mesg_data[MSGMAX];
};
#endif

客户端程序sysv_client.c

#include "svmsg.h"
void client(int ,int);

int main(int argc,char *argv[])
{
   int     msqid;
   if((msqid = msgget(MQ_KEY,0)) == -1)
   {
       perror("megget()");
       exit(-1);
   }
   client(msqid,msqid);
   exit(0);
}

void client(int readfd,int writefd)
{
   size_t len;
   ssize_t n;
   char *ptr;
   struct mymesg mesg;
   printf("Send request to server.\n");
   //set pid to message
   snprintf(mesg.mesg_data,MSGMAX,"%ld",(long)getpid());
   len = strlen(mesg.mesg_data);
   mesg.mesg_data[len] = ' '; //blank
   ptr = mesg.mesg_data+len+1;
   printf("Enter filename: ");
   fgets(ptr,MSGMAX-len,stdin);
   len = strlen(mesg.mesg_data);
   if(mesg.mesg_data[len-1] == '\n')
       len--;
   mesg.mesg_len = len;
   mesg.mesg_type = 1;
   printf("mesg_data is :%s len=%ld\n",mesg.mesg_data, mesg.mesg_len);
   if(msgsnd(writefd,&(mesg.mesg_type),mesg.mesg_len,0) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   //read from IPC,write to standard output
   mesg.mesg_type = getpid();
   while( (n = msgrcv(readfd,&(mesg.mesg_type),MSGMAX,mesg.mesg_type,0))>0)
   {
       write(STDOUT_FILENO,mesg.mesg_data,n);
       putchar('\n');
   }
   if(n == 0 )
   {
       printf("Read file from server is completed.\n");
   }
   if(n == -1)
   {
       perror("msgrcv() error");
       exit(-1);
   }
}

服务器程序sysv_server.c

#include "svmsg.h"
void server(int ,int);
int main(int argc,char *argv[])
{
   int     msqid;
   if((msqid = msgget(MQ_KEY,SVMSG_MODE | IPC_CREAT)) == -1)
   {
       perror("megget()");
       exit(-1);
   }
   server(msqid,msqid);
   exit(0);
}

void server(int readfd,int writefd)
{
   FILE            *fp;
   char            *ptr;
   pid_t           pid;
   ssize_t         n;
   ssize_t         len;
   struct mymesg   mesg;
   printf("Waiting for client......\n");
   for(; ;)
   {
       mesg.mesg_type = 1;
       if((n = msgrcv(readfd,&(mesg.mesg_type),MSGMAX,mesg.mesg_type,0)) == 0)
       {
           printf("pathname missing.\n");
           continue;
       }
       mesg.mesg_data[n] = '\0';
       printf("Received message from client is: %s\n",mesg.mesg_data);
       if ((ptr = strchr(mesg.mesg_data,' ')) == NULL)
       {
           printf("bogus request: %s\n",mesg.mesg_data);
           continue;
       }
       *ptr++ = 0;
       pid = atoi(mesg.mesg_data);
       mesg.mesg_type = pid;
       //open fiel and read data
       if((fp = fopen(ptr,"r")) == NULL)
       {
           printf("open file failed.sent msg to client\n");
           snprintf(mesg.mesg_data+n,sizeof(mesg.mesg_data)-n,": can't open,%s\n",strerror(errno));
           mesg.mesg_len = strlen(ptr);
           memmove(mesg.mesg_data,ptr,mesg.mesg_len);
           if(msgsnd(writefd,&(mesg.mesg_type),mesg.mesg_len,0) == -1)
           {
               perror("msgsnd() error");
               exit(-1);
           }
       }
       else
       {
           printf("open file successed.sent file to client\n");
           while(fgets(mesg.mesg_data,MSGMAX,fp) != NULL)
           {
               mesg.mesg_len = strlen(mesg.mesg_data);
               if(msgsnd(writefd,&(mesg.mesg_type),mesg.mesg_len,0) == -1)
               {
                   perror("msgsnd() error");
                   exit(-1);
               }
           }
           fclose(fp);
       }
       printf("send compelted.\n");
       mesg.mesg_len = 0;
       if(msgsnd(writefd,&(mesg.mesg_type),mesg.mesg_len,0) == -1)
       {
           perror("msgsnd() error");
           exit(-1);
       }
   }
}

程序测试结果如下所示：

例子2：每个客户一个队列，将例子1改成所有用户用一个共同的消息队列向服务器发送消息，给每个客户分配一个消息队列，使得服务器对每个客户进行应答。通信模型如下：

以并发服务器模型编写这个程序，服务器给每个客户fork一个子进程进行处理。程序如下：

公共头文件svmsg.h和svmsg.c：

//svmsg.h file
#ifndef  SVMSG_H
#define  SVMSG_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <errno.h>

#define MSG_R 0400 /* read permission */
#define MSG_W 0200 /* write permission */
#define SVMSG_MODE (MSG_R | MSG_W | MSG_R >>3 | MSG_R >>6)
#define MQ_KEY  1234L
#define MSGMAX  1024
//message structure
struct mymesg
{
   long mesg_len;
   long mesg_type;
   char mesg_data[MSGMAX];
};

ssize_t mesg_send(int id,struct mymesg *mptr);
ssize_t mesg_recv(int id,struct mymesg *mptr);

void    Mesg_send(int id,struct mymesg *mptr);
ssize_t Mesg_recv(int id,struct mymesg *mptr);
#endif

//svmsg.c file
#include "svmsg.h"

ssize_t mesg_send(int id,struct mymesg *mptr)
{
   return (msgsnd(id,&(mptr->mesg_type),mptr->mesg_len,0));
}

ssize_t mesg_recv(int id,struct mymesg *mptr)
{
   ssize_t n;
   n = msgrcv(id,&(mptr->mesg_type),MSGMAX,mptr->mesg_type,0);
   mptr->mesg_len = n;
   return n;
}

void Mesg_send(int id,struct mymesg *mptr)
{
   if(mesg_send(id,mptr) == -1)
   {
       perror("mesg_send() error");
       exit(-1);
   }
}
ssize_t Mesg_recv(int id,struct mymesg *mptr)
{
   ssize_t n;
   do
   {
       n = mesg_recv(id,mptr);
   }while(n==-1 && errno == EINTR);
   if(n == -1)
   {
       perror("mesg_recv() error");
       exit(-1);
   }
   return n;
}

客户端程序如下：

#include "svmsg.h"

void client(int ,int);

int main(int argc,char *argv[])
{
   int    readid,writeid;
   if((writeid = msgget(MQ_KEY,0)) == -1)
   {
       perror("megget()");
       exit(-1);
   }
   if((readid = msgget(IPC_PRIVATE,SVMSG_MODE | IPC_CREAT)) == -1)
   {
       perror("megget()");
       exit(-1);
   }
   client(readid,writeid);
   msgctl(readid,IPC_RMID,NULL);
   exit(0);
}

void client(int readid,int writeid)
{
   size_t len;
   ssize_t n;
   char *ptr;
   struct mymesg mesg;
   printf("Send request to server.\n");
   //set pid to message
   snprintf(mesg.mesg_data,MSGMAX,"%d",readid);
   len = strlen(mesg.mesg_data);
   mesg.mesg_data[len] = ' '; //blank
   ptr = mesg.mesg_data+len+1;
   printf("Enter filename: ");
   fgets(ptr,MSGMAX-len,stdin);
   len = strlen(mesg.mesg_data);
   if(mesg.mesg_data[len-1] == '\n')
       len--;
   mesg.mesg_len = len;
   mesg.mesg_type = 1;
   printf("mesg_data is :%s\n",mesg.mesg_data);
   Mesg_send(writeid,&mesg);
   printf("Send messge to server successed.\n");
   //read from IPC,write to standard output
   while( (n = Mesg_recv(readid,&mesg))>0)
   {
       write(STDOUT_FILENO,mesg.mesg_data,n);
       putchar('\n');
   }
   if(n == 0 )
   {
       printf("Read file from server is completed.\n");
   }
}

服务器程序如下：

#include "svmsg.h"

void server(int ,int);
void sig_child(int signo);

int main(int argc,char *argv[])
{
   int     msqid;
   if((msqid = msgget(MQ_KEY,SVMSG_MODE | IPC_CREAT)) == -1)
   {
       perror("megget()");
       exit(-1);
   }
   server(msqid,msqid);
   exit(0);
}

void server(int readid,int writeid)
{
   FILE            *fp;
   char            *ptr;
   pid_t           pid;
   ssize_t         n;
   ssize_t         len;
   struct mymesg   mesg;
   signal(SIGCHLD,sig_child);
   printf("Waiting for client......\n");
   for(; ;)
   {
       mesg.mesg_type = 1;
       if((n = Mesg_recv(readid,&mesg)) == 0)
       {
           printf("pathname missing.\n");
           continue;
       }
       mesg.mesg_data[n] = '\0';
       printf("Received message from client is: %s\n",mesg.mesg_data);
       if ((ptr = strchr(mesg.mesg_data,' ')) == NULL)
       {
           printf("bogus request: %s\n",mesg.mesg_data);
           continue;
       }
       *ptr++ = 0;
       writeid = atoi(mesg.mesg_data);
       if(fork() == 0)
       {
           //open fiel and read data
           if((fp = fopen(ptr,"r")) == NULL)
           {
               printf("open file failed.sent msg to client\n");
               snprintf(mesg.mesg_data+n,sizeof(mesg.mesg_data)-n,": can't open,%s\n",strerror(errno));
               mesg.mesg_len = strlen(ptr);
               memmove(mesg.mesg_data,ptr,mesg.mesg_len);
               Mesg_send(writeid,&mesg);
           }
           else
           {
               printf("open file successed.sent file to client\n");
               while(fgets(mesg.mesg_data,MSGMAX,fp) != NULL)
               {
                   mesg.mesg_len = strlen(mesg.mesg_data);
                   Mesg_send(writeid,&mesg);
               }
               fclose(fp);
           }
           printf("send compelted.\n");
           mesg.mesg_len = 0;
           Mesg_send(writeid,&mesg);
       }
   }
}

void sig_child(int signo)
{
   pid_t   pid;
   int     stat;
   while ((pid = waitpid(-1,&stat,WNOHANG)) > 0);
   return ;
}

程序测试结果如下：

码农公寓

相关文章