用消息隊列實現Client和Server間的通信


本文主要介紹UNIX操作系統中消息隊列機制、調用方法以及多路復用消息技術，通過應用實例，給出了其在Client和Server間的通信實現。

網絡應用的標準模型是客戶一服務員模型。在金融系統應用軟件開發中，這種模型被廣泛的采用，并且實現這種模型的方法多種多樣。因此，根據不同的業務類型，選擇較好的實現方法至關重要。

在UNIX系統的內部結構中，含有消息機構，即所有的消息都放在內核中，并且它們都有一個相應的消息隊列號。消息機構允許進程發送一個消息到任意其它進程，從而實現系統中進程間的通信。

一、UNIX系統中的消息機構
消息是一個格式化的可變長度的信息單元。它有如下屬性：(1)長整數類型 (2)消息的數據長度 (3)數據。由于消息的長度是可變的，故將消息分為消息首部和消息數據兩部分。在消息首部中，記錄著消息的類型和大小，指向消息數據區的指針，消息隊列的鏈接指針等。每個消息隊列有一個稱為 key的名稱，如同用戶文件描述符一樣，每個消息隊列還有一個消息隊列描述符。此外，在一個系統中，可能有若干個消息隊列，所有消息隊列的頭標組成一個數組。圖1 示出了消息和消息隊列的數據結構。

圖1 消息和消息隊列的數據結構

1. 建立或返回消息隊列描述符
進程可用系統調用megget來建立或返回消息隊列的描述符。該系統調用的語法格式為：
int megget(key,msgflg)
key_t key;
int megflg;
其中，key是消息隊列的名字；msgflg是用戶設置的標志。如果IPC_CREAT表示系統無以key命名的消息隊列，則建立消息隊列標識符；若已存在，則返回消息隊列描述符msgid。

　　對于系統調用，核心將搜索消息隊列頭標數組，確定是否有指定關鍵字的消息隊列。若無，核心將分配一新的隊列結構，并返回給用戶一個消息隊列描述符；否則，它只是檢查消息隊列的許可權之后便返回。
2.消息的發送
進程可用megsnd( )系統調用來發送一個消息，并將它鏈入消息隊列的尾部。該系統調用的語法格式如下：
int msgsnd(msgid,msgp,msgsz,msgflg)
int msgid;
struct msgbuf * msgp;
int msgsz,msgflg;
其中，msgid是由msgget返回的消息隊列描述符；msgp指向包含這條消息的結構，該結構由如下兩個成員組成：
struct msgbuf
{
long mtype; /* 消息類型 */
char mtext[ ]; /* 消息的文本 */
}
msgsz是mtext的字節長度；msgflg規定了當無內存空間來存儲消息時，進程等待還是立即返回。
對于msgsnd( )系統調用，核心檢查消息隊列描述符和許可權是否合法；消息長度是否超過系統規定的長度，若過長，進程睡眠等待出現足夠大的空間，通過檢查后，核心為消息分配消息數據區，并將消息從用戶空間拷貝到消息數據區，分配消息首部，將它鏈入該消息隊列的尾部，在消息首部填寫消息類型，大小以及指向消息數據區的指針，還有修改消息隊列的頭標中的數據。然后喚醒在等待消息到來的隊列中睡眠的進程。
3. 消息的接收
進程可用msgrcv( )系統調用，從消息隊列中讀一條消息，語法格式為：
int msgrcv(msgid,msgp,msgsz,msgtyp,msgflg)
int msgid,msgsz,msgflg;
struct msgbuf * msgp;
long msgtyp;
其中，msgid,msgp,msgsz,msgflg與msgsnd相似，msgtype是規定用戶想讀的消息類型。
對于msgrcv( )系統調用是先由核心檢查消息隊列標識符和許可權，接著根據msgtyp分三種情況處理。
(1) msgtyp=0，核心尋找消息隊列中的第一個消息，并將它返回給調用進程；
(2)msgtyp為正整數，核心返回給類型的第一個消息；
(3)msgtyp為負整數，核心應在其類型值小于或等于msgtyp絕對值的所有消息中，選擇類型最低的第一消息返回。
如果所返回的消息的大小等于或小于用戶請求，核心便將消息正文拷貝到用戶區，再從隊列中刪除該消息，并喚醒睡眠的發送進程；如果消息比用戶要求的大，則系統返回錯誤信息。
4. 消息隊列的操縱
可利用msgctl( )系統調用，來改變消息隊列的屬性，即擁有者，許可權等。其語法格式如下：
int msgctl(msgid,cmd,buf)
int msgid,cmd;
struct msgid_ds * buf;
其中，cmd是規定的命令；buf是用戶緩沖區地址，用戶用它來存放控制參數和查詢結果。命令可分為三類：(1)用于查詢有關消息隊列的情況。(2)用于改變有關消息隊列的屬性。(3)消除消息隊列的標識符。
二、多路復用消息
在客戶-服務員模型中，一個服務員往往對應多個客戶。這時我們可以利用消息的類型參量，讓多個進程把消息放入同一個隊列中，以便消息隊列能夠多路復用。如圖2 所示，我們只要把type置為1，以表示消息是從客戶流向服務員的。如果客戶把它的進程號作為消息的一部分傳遞，那么服務員只要把客戶進程號作為其消息類型，把它的消息發送給客戶進程。每個客戶進程都把msgrcv的參數msgtyp置為其進程號。


圖2 消息隊列的多路復用

三、應用實例
由上面介紹，我們了解了UNIX 操作系統中消息隊列機構及調用方法，為了進一步說明其在Client 和Server 間的通信，我們給出如下應用實例：
1.實例說明
本例主要是Client 端向Server 端提出查詢申請，Server 端通過訪問系統上INFORMIX 數據庫，將所得信息反饋給Client 端。其中，把消息類型設為1表明消息是從Client 端傳到Server 端，消息類型設為2表明消息是從Server 端傳到Client 端。
INFORMIX 中的數據庫dxddb 里的表 dxdtable 數據結構如下：
(1) name char( /* 姓名 */
(2) acct_no char(16) /* 帳號 */
(3) ph_code char( /* 電話號碼 */

2.程序說明

(1)頭文件(文件名：msgg.h)
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define MKEY1 1234L
#define MKEY2 2345L
#define RERMS 0666

(2)公用函數
mesg_recv(id,mesgptr)
int id;
Mesg *mesgptr;
{
    int n;
    n=msgrcv(id,(char *) & (mesgptr->mesg_type),MAXMESGDATA,
    mesgptr->mesg_type,0);
    if((mesgptr->mesg_len=n)<0)
    printf("msgrcv error.\n");
    return(n);
}

mesg_send(id,mesgptr)
int id;
Mesg * mesgptr;
{
    if(msgsnd(id,(char*)&(mesgptr->mesg_type),
    mesgptr->mesg_len,0)!=0)
    printf("msgsnd error\n");
}

(3)Client端程序(文件名：client.c)
#include "msgg.h"
#include <stdio.h>
#define MAXMESGDATA 1024
typedef struct
{
    int mesg_len;
    long mesg_type;
    char mesg_data[1024];
} Mesg;

Mesg mesg;
int i;
main()
{
    int id;
    int sum;
    sum=0;
    for(; {
    if((id=msgget(MKEY1,0))<0)
    printf("client: can not msgget message queue 1.\n");
    else
    printf("CLIENT(%d)-> please send a message: ",sum++);
    client(id);
    }
    if(msgctl(id,IPC_RMID,(struct msqid_ds*)0)<0)
    printf("client: can't PMIN message queue 1.\n");
    exit(0);
}

client(id)
int id;
{
    int n;
    if (fgets(mesg.mesg_data,MAXMESGDATA,stdin)==NULL)
    printf("filename read error.\n");
    n=strlen(mesg.mesg_data);
    if(mesg.mesg_data[n-1]=='\n')
    n--;
    mesg.mesg_data[n]='\0';
    mesg.mesg_len=n;
    mesg.mesg_type=1L;
    mesg_send(id,&mesg);
    printf("client: this message has been sent to server. \nthe message from server is: \n\n");
    mesg.mesg_type=2L;
    while((n=mesg_recv(id,&mesg))>0)
    if(write(1,mesg.mesg_data,n)!=n)
    printf("data write error.\n");
    if (n>0)
    printf("data read error.\n");
}

(4)Server端程序(文件名：server.ec)
#include "msgg.h"
#include <stdio.h>
#define MESGHDRSIZE 1024
$typedef struct
{
    char name[9];
    char acct_no[17];
    char ph_code[8];
} st;
$typedef struct
{
    int mesg_len;
    long mesg_type;
    char mesg_data[8];
    st p_code;
} Mesg;
$typedef struct
{
    int mesg_len;
    long mesg_type;
    char mesg_error[20];
} Mesg1;
$include sqltypes.h;
$include sqlca;
$Mesg mesg;
$Mesg1 mesg1;
$char mm[9];
main()
{
    int id;
    int sum;
    sum=0;
    for(;
    {
        if((id=msgget(MKEY1,RERMS|IPC_CREAT))<0)
        printf("server: can not get message queue 1.\n");
        else
        {
            printf("server: waitting for server.... ");
            printf("%d \n",sum++);
            server(id);
        }
    }
}

server(id)
int id;
{
    int n,filefd;
    $database dxddb;
    mesg.mesg_type=1L;
    if((n=mesg_recv(id,&mesg))<=0)
    printf("serve:filename read error\n");
    mesg.mesg_data[n]='\0';
    strcpy(mm,mesg.mesg_data);
    printf("this message is :%s \n",mm);
    mesg.mesg_type=2L;
    $select * into $mesg.p_code
    from dxdtable
    where ph_code = $mm;
    if(sqlca.sqlcode==0)
    {
        printf("this name is :%s\n",mesg.p_code.name);
        printf("begin to send this message to client.\n");
        mesg.mesg_len=sizeof(mesg);
        mesg_send(id,&mesg);
        mesg.mesg_len=0;
        mesg_send(id,&mesg);
    }
    else
    {
        printf("this phone number is not exist ! \n");
        strcpy(mesg1.mesg_error,"there are not this number !");
        mesg1.mesg_type=2L;
        mesg1.mesg_len=strlen(mesg1.mesg_error);
        mesg_send(id,&mesg1);
        mesg1.mesg_len=0;
        mesg_send(id,&mesg1);
    }
}