VxWorks任务间使用消息队列通信

/* includes */
#include "vxWorks.h"
#include "taskLib.h"
#include "msgQLib.h"
#include "sysLib.h"
#include "stdio.h"

 

/* globals */
#define  CONSUMER_TASK_PRI          99   /* Priority of the consumer task */
#define  PRODUCER_TASK_PRI          98   /* Priority of the producer task */
#define  TASK_STACK_SIZE          5000   /* stack size for spawned tasks */

struct msg   /* data structure for msg passing */
{                            
    int tid;                         /* task id */        
    int value;                       /* msg value */
};

LOCAL MSG_Q_ID msgQId;                   /* message queue id */
LOCAL int numMsg = 8;                    /* number of messages */
LOCAL BOOL notDone;                      /* Flag to indicate the completion of this demo */

 

/* function prototypes */
LOCAL STATUS producerTask ();           /* producer task */
LOCAL STATUS consumerTask ();           /* consumer task */

 

/* user entry */
STATUS msgQDemo()
{
    notDone = TRUE;  /* initialize the global flag */
    /* Create the message queue*/
    if ((msgQId = msgQCreate (numMsg, sizeof (struct msg), MSG_Q_FIFO))
        == NULL)
    {
        perror ("Error in creating msgQ");
        return (ERROR);
    }
     /* Spwan the producerTask task */
    if (taskSpawn ("tProducerTask", PRODUCER_TASK_PRI, 0, TASK_STACK_SIZE,
                   (FUNCPTR) producerTask, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
                    == ERROR)
    {
        perror ("producerTask: Error in spawning demoTask");
        return (ERROR);
    }
   
    /* Spwan the consumerTask task */
    if (taskSpawn ("tConsumerTask", CONSUMER_TASK_PRI, 0, TASK_STACK_SIZE,
                   (FUNCPTR) consumerTask, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
               == ERROR)
    {
        perror ("consumerTask: Error in spawning demoTask");
        return (ERROR);
    }
   
    /* polling is not recommended. But used to make this demonstration simple*/
    while (notDone)
        taskDelay (sysClkRateGet ());
       
    if (msgQDelete (msgQId) == ERROR)
    {
        perror ("Error in deleting msgQ");
        return (ERROR);
    }
   
    return (OK);
}

 

/* producerTask ：发送消息给consumerTask */
STATUS producerTask (void)
{
    int count;
    int value;
    struct msg producedItem;           /* producer item - produced data */
    printf ("producerTask started: task id = %#x \n", taskIdSelf ());
   
    /* Produce numMsg number of messages and send these messages */
    for (count = 1; count <= numMsg; count++)
    {
        value = count * 10;   /* produce a value */
       
        /* Fill in the data structure for message passing */
        producedItem.tid = taskIdSelf ();
        producedItem.value = value;
       
        /* Send Messages */
        if ((msgQSend (msgQId, (char *) &producedItem, sizeof (producedItem),
                       WAIT_FOREVER,  MSG_PRI_NORMAL)) == ERROR)
        {
            perror ("Error in sending the message");
            return (ERROR);
        }
        else
            printf ("ProducerTask: tid = %#x, produced value = %d \n",
                                                      taskIdSelf (), value);
    }
   
    return (OK);
}

 

/* consumerTask：获取（消费）消息 */
STATUS consumerTask (void)
{
    int count;
    struct msg consumedItem;           /* consumer item - consumed data */
    printf ("\n\nConsumerTask: Started -  task id = %#x\n", taskIdSelf());
    /* consume numMsg number of messages */
    for (count = 1; count <= numMsg; count++)
    {
        /* Receive messages */
        if ((msgQReceive (msgQId, (char *) &consumedItem,
                          sizeof (consumedItem), WAIT_FOREVER)) == ERROR)
        {
            perror ("Error in receiving the message");
            return (ERROR);
        }
        else
            printf ("ConsumerTask: Consuming msg of value %d from tid = %#x\n",
                            consumedItem.value, consumedItem.tid);
    }
    notDone = FALSE;   /* set the global flag to FALSE to indicate completion*/
    return (OK);
}

 

程序运行输出：
producerTask started: task id = 0x1010f20
ProducerTask: tid = 0x1010f20, produced value = 10
ProducerTask: tid = 0x1010f20, produced value = 20
ProducerTask: tid = 0x1010f20, produced value = 30
ProducerTask: tid = 0x1010f20, produced value = 40
ProducerTask: tid = 0x1010f20, produced value = 50
ProducerTask: tid = 0x1010f20, produced value = 60
ProducerTask: tid = 0x1010f20, produced value = 70
ProducerTask: tid = 0x1010f20, produced value = 80

ConsumerTask: Started -  task id = 0x100c840
ConsumerTask: Consuming msg of value 10 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 20 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 30 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 40 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 50 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 60 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 70 from tid = 0x1010f20
ConsumerTask: Consuming msg of value 80 from tid = 0x1010f20

VxWorks主要提供如下API进行消息队列的创建、读取和传递：
msgQCreate( )：创建斌初始化一个消息队列，函数原型为：
MSG_Q_ID  msgQCreate
(
    int    maxMsgs,         /*队列所能容纳的最大消息数目*/
    int    maxMsgLength,    /*每一消息的最大长度*/
    int    options          /*消息入列方式*/
);

msgQDelete( )：终止并释放一个消息队列，函数原型为：
STATUS       msgQDelete
(
    MSG_Q_ID  msgQId  /*要删除的消息队列ID号*/
);
msgQSend( )：发送一个消息到消息队列，函数原型为：
STATUS       msgQSend
(
    MSG_Q_ID     msgQId,        /*所发向的消息队列名*/
    char *       buffer,       /*消息包所在缓冲区指针*/
    UINT         nBytes,         /*消息包长度*/
    int          timeout,  /*等待的时间长度*/
    int           priority   /*优先级*/
);
msgQReceive( )：从消息队列接受一个消息，函数原型为：
int msgQReceive
(
    MSG_Q_ID  msgQId, /*接收消息的消息队列ID号*/
    char *     buffer,     /*接收消息的缓冲区指针*/
    UINT            maxNBytes,  /*缓冲区长度*/
    int           timeout   /*等待时间*/
);