当前位置：首页 > news >正文

消息队列缓存，以蓝牙消息服务为例

news 2025/7/19 13:23:18

前言

消息队列缓存，支持阻塞、非阻塞模式；支持协议、非协议模式

可自定义消息结构体数据内容

使用者只需设置一些宏定义、调用相应接口即可

这里我用蓝牙消息服务举例

有纰漏请指出，转载请说明。

学习交流请发邮件 1280253714@qq.com

原理讲解

队列采用"生产者-消费者"模式，

当接收数据时，生产者即为外部输入，输入的消息入列；消费者即为消息处理函数，处理消息前先出列，然后消费者处理出列的消息

当发送数据时，生产者为消息产生的任务，该任务将产生的消息入列；消费者为消息发送任务，该任务将待发送的消息出列并发送。

接口

接收消息接口

以蓝牙接收消息服务为例

接收消息处理接口

这里我开了一个定时任务来处理消息

待发送消息装载接口

由任务产生消息并调用装载函数，以下提供透明传输和协议传输接口

待发送消息发送接口

以蓝牙发送消息服务为例，这里我先进行出列判断，如果数据成功出列，则将消息通过蓝牙的内核发送出去

代码

msg.h

#ifndef _MSG_H
#define _MSG_H
#include "includes.h"#define MsgProtocol	1	//是否使用协议
#define MsgBlock	1	//消息是否阻塞，即当队列已满时，是否继续入列，//入列会导致前面接收到且未来得及处理的数据被覆盖						 #define QueueMaxSize 5	//队列最大存放的数据个数
#define MsgMaxSize 64	//单帧数据最大长度									 
#define MsgPrefix 0xA5	
#define MsgStuffix 0x5A
#define MsgIndexPrefix 0
#define MsgIndexLen 1
#define MsgIndexCmdId 2
#define MsgIndexCmd 3
#define MsgIndexDataBase 4typedef enum			//队列当前状态
{eQueueEmpty,		//队列为空eQueueNormal,		//队列不为空eQueueFull,			//队列已满
} QueueState_E;			typedef enum 			//出列的数据情况
{eDataEmpty,			//数据为空eDataNormal,		//数据不为空
} DeQueueState_E;		typedef struct			//数据格式
{u8 u8Len;u8 szData[MsgMaxSize];
} MSG_S;										 typedef struct 					//队列结构体
{u32 msgCnt;		u32 msgProcCnt;MSG_S dataEnQueue;MSG_S dataDeQueue;MSG_S data[QueueMaxSize];QueueState_E state;			//队列当前状态u8 front;					//队头u8 rear;					//队尾u8 size;					//队列大小
}MSG_ATTR_S;typedef struct					//蓝牙消息结构体
{MSG_ATTR_S stRx;MSG_ATTR_S stTx;
}BLE_MSG_S; QueueState_E MsgQueueStateDetermine(MSG_ATTR_S *queue);
void MsgEnQueue(MSG_ATTR_S *queue, MSG_S *data);
DeQueueState_E MsgDeQueue(MSG_ATTR_S *queue);
void MsgRxDataProc(void);
int MsgProcCrc(u8 *pRxData);
void LoadTransparentMsg(MSG_ATTR_S *queue, MSG_S *txMsg);
void LoadProtocalMsg(MSG_ATTR_S *queue, MSG_S *txMsg);
void BleMsgInit(void);extern MSG_ATTR_S stRxQueue;
extern BLE_MSG_S stBle;		#endif

msg.c

#include "includes.h"BLE_MSG_S stBle;int MsgProcCrc(u8 *pRxData)
{u8 i = 0;u8 crc = 0;u8 size = pRxData[MsgIndexLen];if ((pRxData[MsgIndexPrefix] == MsgPrefix) && (pRxData[size - 1] == MsgStuffix)){for (i = 1; i <= size - 3; i++){crc += pRxData[i];}if (crc == pRxData[size - 2]){return 0;}}return -1;
}QueueState_E MsgQueueStateDetermine(MSG_ATTR_S  *queue)
{
#if MsgBlockif (queue->size == 0){return eQueueEmpty;}else if (queue->size == QueueMaxSize){return eQueueFull;}else{return eQueueNormal;}	
#elseif (queue->msgCnt == 0){return eQueueEmpty;}else if ( queue->msgCnt > 0){return eQueueNormal;}
#endif		
}void MsgEnQueue(MSG_ATTR_S *queue, MSG_S *data)
{
#if MsgBlockif (queue->size == QueueMaxSize){return;}
#endif	queue->dataEnQueue = *data;queue->data[queue->rear] = queue->dataEnQueue;queue->size++;queue->msgCnt++;queue->rear++;if (queue->rear == QueueMaxSize){queue->rear = 0;}queue->state = MsgQueueStateDetermine(queue);}DeQueueState_E MsgDeQueue(MSG_ATTR_S *queue)
{if (queue->size == 0){return eDataEmpty;}queue->dataDeQueue = queue->data[queue->front];memset(&queue->data[queue->front], 0, sizeof(queue->data[queue->front]));queue->size--;queue->front++;if (queue->front == QueueMaxSize){queue->front = 0;}queue->state = MsgQueueStateDetermine(queue);return eDataNormal;
}void MsgRxDataProc(void)
{if (stBle.stRx.state != eQueueEmpty){if (MsgDeQueue(&stBle.stRx) != eDataEmpty){stBle.stRx.msgProcCnt++;}}
}void LoadTransparentMsg(MSG_ATTR_S *queue, MSG_S *txMsg)
{MsgEnQueue(queue, txMsg);
}void LoadProtocalMsg(MSG_ATTR_S *queue, MSG_S *txMsg)
{u8 i;u8 checkSum = 0;MSG_S tmpMsg;memset(&tmpMsg, 0, sizeof(MSG_S));for (i = 0; i < txMsg->u8Len; i++){checkSum += txMsg->szData[i];tmpMsg.szData[i+2] = txMsg->szData[i];        }tmpMsg.szData[MsgIndexPrefix] = MsgPrefix;tmpMsg.szData[txMsg->u8Len + 3] = MsgStuffix;tmpMsg.szData[MsgIndexLen] = txMsg->u8Len + 4;checkSum += tmpMsg.szData[MsgIndexLen];tmpMsg.szData[txMsg->u8Len + 2] = checkSum;tmpMsg.u8Len = txMsg->u8Len + 4;MsgEnQueue(queue, &tmpMsg);
}void BleMsgInit(void)
{memset(&stBle, 0, sizeof(BLE_MSG_S));
}