c实现自定义消息队列实例源码

C实现消息队列基于现在的很多实时操作系统都舍弃了消息队列的接口，导致我们在资源共享和传递时比较费力，尤其是在CPU上裸跑的情况，总是要为了优化数据处理，而花上很长时间，而且感觉每次都是做重复的事情。这里我把它终结出来以便后面直接参考消化，避免重复造车轮。当然这个工程只是个人拙作，肯定有不足的地方，还请大家不吝赐教。好废话不多说，上实例（因为代码太多只贴头文件和main.c）有需要可以免费下...

希哈科技

610人浏览 · 2019-03-28 16:55:31

希哈科技 · 2019-03-28 16:55:31 发布

C实现消息队列

基于现在的很多实时操作系统都舍弃了消息队列的接口，导致我们在资源共享和传递时比较费力，尤其是在CPU上裸跑的情况，总是要为了优化数据处理，而花上很长时间，而且感觉每次都是做重复的事情。这里我把它终结出来以便后面直接参考消化，避免重复造车轮。当然这个工程只是个人拙作，肯定有不足的地方，还请大家不吝赐教。

好废话不多说，上实例（因为代码太多只贴头文件和main.c）有需要可以免费下载我的工程代码。。。

网盘：http://pan.baidu.com/s/1mhBCIrI

密码：yx4b

list.h文件

/*
* list.h
* Created on: 2017年9月5日
* Author: jobschu
*/

#ifndef __LIST_H__
#define __LIST_H__

struct list {
   struct list *next;
   struct list *prev;
};
struct list_x {
   struct list *next;
   struct list *prev;
};
/*init list*/
void list_init(struct list *list);
/*free list*/
int list_empty(struct list *list);

/*insert before link */
void list_insert(struct list *link, struct list *new_link);
/*insert before link */
void list_append(struct list *list, struct list *new_link);
/*insert after link*/
void list_prepend(struct list *list, struct list *new_link);
/*delete this link*/
void list_remove(struct list *link);

#define list_entry(link, type, member) \
   ((type *)((char *)(link)-(unsigned long)(&((type *)0)->member)))

/*get list head*/
#define list_head(list, type, member)       \
   list_entry((list)->next, type, member)

#define list_tail(list, type, member)       \
   list_entry((list)->prev, type, member)

#define list_next(elm, member)                   \
   list_entry((elm)->member.next, typeof(*elm), member)

#define list_for_each_entry(pos, list, member)           \
   for (pos = list_head(list, typeof(*pos), member);   \
   &pos->member != (list);               \
   pos = list_next(pos, member))

#endif

msg.h
/*
* msg.h
* Created on: 2017年9月5日
* Author: jobschu
*/
#ifndef __MSG_H__
#define __MSG_H__

#include "list.h"

#define TID_MSG 0

struct msg {
    struct list nodeHead;/*消息队列的头*/
    unsigned int type;
   unsigned int len;
   //unsigned int lock;
   void *pdata;
};
typedef struct msg msg_t;
typedef struct msg * pMsg_t;

/*新建消息资源*/
pMsg_t msg_new(unsigned int type,unsigned char *data,int len);
/*消息队列在取得消息数据后，必须释放资源*/
void msg_free(pMsg_t pMsg);
#endif

msgQueue.h

/*
* msgQueue.h
* Created on: 2017年9月5日
* Author: jobschu
*/

#ifndef __MSG_QUEUE_H__
#define __MSG_QUEUE_H__

struct msgQueue {
    struct list queueHead;   /*消息队列链表头*/
    unsigned int length;   /*消息队列长度*/
   unsigned char lock;       /*消息队列锁，防止在多线程里的互斥操作*/
};
typedef struct msgQueue msgQueue_t;
typedef struct msgQueue * pMsgQueue_t;

/*创建消息队列成功返回消息队列handle 失败返回NULL*/
pMsgQueue_t msgQueue_create(void);

/*从消息队列中获悉消息*/
pMsg_t msgQueue_getMsg(pMsgQueue_t q);

/*将消息添加到队列尾部*/
int msgQueue_putMsg(pMsgQueue_t q,pMsg_t m);
/*判断消息队列是否为空为空返回1 否则返回0*/
int msgQueue_isEmpty(pMsgQueue_t q);
/*清除消息队列*/
int msgQueue_delete(pMsgQueue_t q);

#endif

文件main.c
/*
* main.c
* Created on: 2017年9月5日
* Author: jobschu
*/
#include <stdio.h>
#include <stdlib.h>
#include <sys/ioctl.h>

#include <sys/types.h>
#include <sys/stat.h>

#include <errno.h>

#include <fcntl.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>

#include "msg.h"
#include "msgQueue.h"

static pMsgQueue_t youQueue;
static pMsgQueue_t myQueue;

static pMsg_t myMsg;
static unsigned char buff[][10]={
       "1111111111","2222222222","3333333333","4444444444",
       "5555555555","6666666666","7777777777","8888888888"
};
/*测试一实现单个消息队列发送单个消息队列接收测试*/
int test_0(void){
   int num = 0,i=0;
   printf("only for test!\n");
   /*创建消息队列*/
   myQueue = msgQueue_create();
   if(myQueue == NULL){
       printf("new msg queue err!\n");
       return -1;
   }
   printf("add msg to queue...\n");
   /*获取要模拟发送的消息长度*/
   num = sizeof(buff)/sizeof(buff[0]);

   for(i=0;i<num;i++){
       /*新建消息*/
       myMsg = msg_new(0,buff[i],sizeof(buff[i]));
       /*将新建的消息推送到消息队列*/
       if(msgQueue_putMsg(myQueue,myMsg)){
           printf("put Msg err:[%d]\n",i);
       }
       //printf("add msg:%s to queue\n",buff[i]);
   }
   printf("get msg from queue...\n");


   while(!msgQueue_isEmpty(myQueue)){
       myMsg = msgQueue_getMsg(myQueue);
       if(myMsg == NULL||myMsg->len <=0 ||myMsg->pdata == NULL){
           continue;
       }
       printf("get msg:len:[%d],data:%s\n",myMsg->len,myMsg->pdata);
       /*将获取到的消息使用后释放资源*/
       msg_free(myMsg);
   }
   printf("main end!\n");
   /*删除消息队列*/
   msgQueue_delete(myQueue);
   return 0;
}

/*测试方法二通过主线程发送消息到消息队列，线程用来异步接收消息处理消息*/
static void *thread(void *str)
{
   int i = 0;
   unsigned char RecvBuff[256];
   unsigned int RecvLen = 0;
   while(1)
   {
       /*获取消息队列中的消息*/
       if(!msgQueue_isEmpty(myQueue)){
           myMsg = msgQueue_getMsg(myQueue);

           if((myMsg != NULL) && (myMsg->len > 0) && (myMsg->pdata != NULL)){
               printf("get msg:len:[%d],data:%s\n",myMsg->len,myMsg->pdata);
               /*释放掉消息占有的空间*/
               msg_free(myMsg);
           }
       }
       usleep(100000);//100ms
   }
}

int test_1(void){

   int num = 0,i=0;
   pthread_t pth;

   printf("start the man process!\n");
   /*创建消息队列*/
   myQueue = msgQueue_create();
   if(myQueue == NULL){
       printf("new msg queue err!\n");
       return -1;
   }

   printf("create new pthread!\n");
   /*创建接收消息队列的线程*/
   pthread_create(&pth,NULL,thread,NULL);

   printf("add msg to queue...\n");
   /*获取要模拟发送的消息长度*/
   num = sizeof(buff)/sizeof(buff[0]);

   for(i=0;i<num;i++){
       /*新建消息*/
       myMsg = msg_new(0,buff[i],sizeof(buff[i]));
       /*将新建的消息推送到消息队列*/
       if(msgQueue_putMsg(myQueue,myMsg)){
           printf("put Msg err:[%d]\n",i);
       }
       usleep(3000000);//1000ms=1s
   }
}

/*创建两个并行的消息队列，交叉利用测试功能*/
int test_2(void){
   int num = 0,i=0;
   printf("only for test!\n");
   /*创建消息队列*/
   myQueue = msgQueue_create();
   if(myQueue == NULL){
       printf("new msg queue err!\n");
       return -1;
   }
   /*创建消息队列*/
   youQueue = msgQueue_create();
   if(youQueue == NULL){
       printf("new msg queue err!\n");
       return -1;
   }

   printf("add msg to myQueue...\n");
   /*获取要模拟发送的消息长度*/
   num = sizeof(buff)/sizeof(buff[0]);

   for(i=0;i<num/2;i++){
       /*新建消息*/
       myMsg = msg_new(0,buff[i],sizeof(buff[i]));
       /*将新建的消息推送到消息队列*/
       if(msgQueue_putMsg(myQueue,myMsg)){
           printf("put Msg err:[%d]\n",i);
       }
   }
   printf("add msg to youQueue...\n");
   for(i=num/2;i<num;i++){
       /*新建消息*/
       myMsg = msg_new(0,buff[i],sizeof(buff[i]));
       /*将新建的消息推送到消息队列*/
       if(msgQueue_putMsg(youQueue,myMsg)){
           printf("put Msg err:[%d]\n",i);
       }
   }

   printf("get youQueue from queue...\n");
   /*将消息从消息队列中读取出来*/
   while(!msgQueue_isEmpty(youQueue)){
       myMsg = msgQueue_getMsg(youQueue);
       if(myMsg == NULL||myMsg->len <=0 ||myMsg->pdata == NULL){
           continue;
       }
       printf("get youQueue:len:[%d],data:%s\n",myMsg->len,myMsg->pdata);
       msg_free(myMsg);
   }
   printf("get myQueue from queue...\n");
   /*将消息从消息队列中读取出来*/
   while(!msgQueue_isEmpty(myQueue)){
       myMsg = msgQueue_getMsg(myQueue);
       if(myMsg == NULL||myMsg->len <=0 ||myMsg->pdata == NULL){
           continue;
       }
       printf("get myQueue:len:[%d],data:%s\n",myMsg->len,myMsg->pdata);
       msg_free(myMsg);
   }

   printf("main end!\n");
   /*删除消息队列*/
   msgQueue_delete(myQueue);
   msgQueue_delete(youQueue);
   return 0;
}

int main(int argc,char **argv)
{
   //return test_0();
   //return test_1();
   return test_2();
}