linux中进程内部消息队列通信代码以及说明

青旅半醒 2022-03-11 16:40 55阅读 0赞

#include <stdio.h>  
    //使用pthread头文件进行创建线程，和sem_wait等相关操作，
    //注意编译时加上-lpthread选项，以调用静态链接库。因为pthread并非Linux系统的默认库。
    #include <pthread.h>
    #include <semaphore.h>  
    #include <unistd.h>  
    #include <string.h>  
    
    #define QUEUE_SIZE 50
    #define GO_HOME 1
    #define THREAD1 11
    #define THREAD2 12
    #define THREAD3 13
    #define HANDLER 21
    
    
    //定义消息头的结构体
    typedef struct Msg_Hdr_s{
    	unsigned int msg_type;//代表队列中每一个消息的类型，通过类型可以判断出来发送的是什么消息
    	unsigned int msg_len;//定长消息的长度
    	unsigned int msg_src;//发送消息的源地址
    	unsigned int msg_dst;//发送消息的目的地址
    }Msg_Hdr_t;
    
    //消息结构体
    typedef struct Msg_s{
    	Msg_Hdr_t hdr;//消息头
    	unsigned int data[100];//定长的消息数据
    }Msg_t;
    //定义队列结构体
    typedef struct Queue_s
    {
    	int head;
    	int rear;
    	sem_t sem;
    	Msg_t data[QUEUE_SIZE];//QUEUE_SIZE是循环队列的长度
    }Queue_t;
    
    Queue_t MsgQueue;  
      
    int MsgQueueInit(Queue_t* Q)  
    {  
        if(!Q)  
        {  
            printf("Invalid Queue!\n");  
            return -1;  
        }  
        Q->rear = 0;  
        Q->head = 0;  
    	//初始化线程，进程函数
    	//    sem_init函数是Posix信号量操作中的函数。sem_init() 初始化一个定位在 sem 的匿名信号量。value 参数指定信号量的初始值。 pshared 参数指明信号量是由进程内线程共享，还是由进程之间共享。如果 pshared 的值为 0，那么信号量将被进程内的线程共享，并且应该放置在这个进程的所有线程都可见的地址上(如全局变量，或者堆上动态分配的变量)。
    	//    如果 pshared 是非零值，那么信号量将在进程之间共享，并且应该定位共享内存区域(见 shm_open(3)、mmap(2) 和 shmget(2))。因为通过 fork(2) 创建的孩子继承其父亲的内存映射，因此它也可以见到这个信号量。所有可以访问共享内存区域的进程都可以用 sem_post(3)、sem_wait(3) 等等操作信号量。初始化一个已经初始的信号量其结果未定义。
    	//    返回值
    	//    sem_init() 成功时返回 0；错误时，返回 -1，并把 errno 设置为合适的值。
    	//    用下面一组函数（系统调用）来实现。
    	//
    	//    int sem_init(sem_t *sem,int pshared,unsigned int value);
    	//    int sem_destroy(sem_t *sem);
    	//    int sem_wait(sem_t *sem);
    	//    int sem_trywait(sem_t *sem);
    	//    int sem_post(sem_t *sem);
    	//    int sem_getvalue(sem_t *sem);
        sem_init(&Q->sem, 0, 1);
    
        return 0;      
    }  
      
    int MsgDeQueue(Queue_t* Q, Msg_t* msg)  
    {  
        if(!Q)  
        {  
            printf("Invalid Queue!\n");  
            return -1;  
        }  
        if(Q->rear == Q->head) //only one cosumer,no need to lock head  
        {  
            printf("Empty Queue!\n");  
            return -1;  
        }  
        memcpy(msg, &(Q->data[Q->head]), sizeof(Msg_t));  
        Q->head = (Q->head+1)%QUEUE_SIZE;  //循环队列，不懂的可以去查看数据结构相关的知识
        return 0;         
    }  
      
    int MsgEnQueue(Queue_t* Q, Msg_t* msg)  
    {  
        if(Q->head == (Q->rear+1)%QUEUE_SIZE)  
        {  
            printf("Full Queue!\n");  
            return -1;  
        }  
    //学过操作系统的人应该知道，信号量互斥时的相关原语PV操作，不懂的自行百度查看操作系统相关知识。
    //    sem_post函数（函数原型  int sem_post(sem_t *sem);）
    //    作用是给信号量的值加上一个“1”。 当有线程阻塞在这个信号量上时，调用这个函数会使其中一个线程不在阻塞，选择机制是有线程的调度策略决定的。
    //    sem_wait函数（函数原型  int sem_wait(sem_t * sem);）
    //    它的作用是从信号量的值减去一个“1”，但它永远会先等待该信号量为一个非零值才开始做减法。
    
        sem_wait(&Q->sem);  
        memcpy(&(Q->data[Q->rear]), msg, sizeof(Msg_t));  
        Q->rear = (Q->rear+1)%QUEUE_SIZE;  
        sem_post(&Q->sem);  
        return 0;  
    }  
      
    void msg_printer(Msg_t* msg)  
    {  
        if(!msg)  
        {  
            return;  
        }  
        printf("%s: I have recieved a message!\n", __FUNCTION__);  
        printf("%s: msgtype:%d   msg_src:%d  dst:%d\n\n",__FUNCTION__,msg->hdr.msg_type,msg->hdr.msg_src,msg->hdr.msg_dst);  
      
    }  
      
    int msg_send()  
    {  
      
        Msg_t msg;  
        msg.hdr.msg_type = GO_HOME;  
        msg.hdr.msg_src = THREAD1;  
        msg.hdr.msg_dst = HANDLER;  
        return MsgEnQueue((Queue_t*)&MsgQueue, &msg);      
      
    }  
      
    void msg_handler()  
    {  
        sleep(5);  //let's wait 5s when starts  
        while(1)  
        {  
            Msg_t msg;  
            memset(&msg, 0 ,sizeof(Msg_t));  
            int res = MsgDeQueue((Queue_t*)&MsgQueue, &msg);  
            if(res != 0)  
            {  
                sleep(10);  
                continue;  
            }  
            msg_printer(&msg);  
            sleep(1);  
        }  
    }  
    
    void msg_sender1()  
    {  
        int i = 0;  
        while(1)  
        {  
            if(i > 10)  
            {  
                i = 0;  
            }  
            Msg_t msg;  
            msg.hdr.msg_type = i++;  
            msg.hdr.msg_src = THREAD1;  
            msg.hdr.msg_dst = HANDLER;  
            MsgEnQueue((Queue_t*)&MsgQueue, &msg);  
            printf("%s: Thread1 send a message!\n",__FUNCTION__);  
            sleep(1);  
        }  
    }  
      
    void msg_sender2()  
    {  
        int i = 0;  
        while(1)  
        {  
            if(i > 10)  
            {  
                i = 0;  
            }  
            Msg_t msg;  
            msg.hdr.msg_type = i++;  
            msg.hdr.msg_src = THREAD2;  
            msg.hdr.msg_dst = HANDLER;  
            MsgEnQueue((Queue_t*)&MsgQueue, &msg);  
            printf("%s: Thread2 send a message!\n",__FUNCTION__);  
            sleep(1);  
        }  
    }  
      
    void msg_sender3()  
    {  
        int i = 0;  
        while(1)  
        {  
            if(i > 10)  
            {  
                i = 0;  
            }  
            Msg_t msg;  
            msg.hdr.msg_type = i++;  
            msg.hdr.msg_src = THREAD3;  
            msg.hdr.msg_dst = HANDLER;  
            MsgEnQueue((Queue_t*)&MsgQueue, &msg);  
            printf("%s: Thread3 send a message!\n",__FUNCTION__);  
            sleep(1);  
        }  
    }
    int main(int argc, char* argv[])
    {
        int ret;
        pthread_t thread1_id;
        pthread_t thread2_id;
        pthread_t thread3_id;
        pthread_t handler_thread_id;
    
        ret = MsgQueueInit((Queue_t*)&MsgQueue);
        if(ret != 0)
        {
            return -1;
        }
    
        if(pthread_create(&handler_thread_id, NULL, (void*)msg_handler, NULL))
        {
            printf("create handler thread fail!\n");
            return -1;
        }
    
    
        if(pthread_create(&thread1_id, NULL, (void*)msg_sender1, NULL))
        {
            printf("create thread1 fail!\n");
            return -1;
        }
    
        if(pthread_create(&thread2_id, NULL, (void*)msg_sender2, NULL))
        {
            printf("create thread2 fail!\n");
            return -1;
        }
    
        if(pthread_create(&thread3_id, NULL, (void*)msg_sender3, NULL))
        {
            printf("create thread3 fail!\n");
            return -1;
        }
        while(1)
        {
            sleep(1);
        }
        return 0;
    }