消息队列应用

client.c

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

struct mymsgbuf{
        long mtype;
        float num_data[10];
};
int main(int argc, char *argv[]){
        struct mymsgbuf buf;
        int msgid;
        int i;
        if((msgid = msgget(0x1234, 0666|IPC_CREAT)) < 0)
        {
            fprintf(stderr, "open msg %x failed.\n", 0x1234);
            return 0;
        }
        memset(&buf, 0, sizeof(buf));
        buf.mtype = atoi(argv[1]);
        srand(atoi(argv[1]) + time(NULL));
        for(i=0; i < 10; i++)
            buf.num_data[i] = 0+1.0*(rand()%RAND_MAX)/RAND_MAX *(1-0);

        msgsnd(msgid, &buf, sizeof(buf.num_data),0);
        usleep(400000);
        msgrcv(msgid, &buf, sizeof(buf.num_data), buf.mtype, 0);
        printf("average is %f\n", buf.num_data[0]);
        return 0;
}

server.c
#include <sys/msg.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <stdio.h>
#include <signal.h>
#include <sys/errno.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#define MSG_NUM 4

extern int errno;
int msgid;
struct mymsgbuf{
    long mtype;
    float num_data[10];
};

void close_fun(int signo)
{
        if(SIGINT == signo)
        {
                msgctl(msgid,IPC_RMID,NULL);
                exit(0);
        }
}

float col_num_avr(float *arr, int num)
{
    int i=0;
    float tmp = 0.0;
    for(i=0; i < num; i++) {
        printf("%f ", arr[i]);
        tmp += arr[i];
    }
    printf("\n");
    return tmp/num;
}
void *deal_num_data(void *arg)
{
    struct mymsgbuf *msg = (struct mymsgbuf *)arg;
    struct mymsgbuf buf;
    float avr = col_num_avr(msg->num_data, 10);
    buf.num_data[0] = avr;
    buf.mtype = msg->mtype;
    msgsnd(msgid, &buf, sizeof(buf.num_data),0);
    return NULL;
}

int main()
{
    struct mymsgbuf *buf[MSG_NUM];
    int i;
    signal(SIGINT, close_fun);
    for(i=0; i < MSG_NUM; i++)
        buf[i] = malloc(MSG_NUM * sizeof(struct mymsgbuf));
    int ret;
    pthread_t pid;
    if((msgid = msgget(0x1234, 0666|IPC_CREAT)) < 0)    {
        fprintf(stderr, "open msg %X failed.\n", 0x1234);
        return 0;
    }
    for(i=0; i < MSG_NUM; i++) {
        buf[i]->mtype=i+1;
        msgrcv(msgid, buf[i], sizeof(buf[i]->num_data), buf[i]->mtype, 0);
        pthread_create(&pid, NULL, deal_num_data, buf[i]);
        usleep(100000);
    }
    return 0;
}

两个程序，客户端生成随机10个float数，然后服务端接受，并返回给client平均值，
执行
./server
seq 1 4 | xargs -P 0 -n 1 ./client