基于RT-Thread的RoboMaster电控框架（六）

背景

使用的开发板为大疆的 RoboMaster-C 型开发板，基础工程为 rt-thread>bsp>stm32f407-robomaster-c

MSG模块开发

MSG 模块主要应用于应用层线程间通讯，实现一种发布者发布话题，订阅者订阅话题的通讯方式。接下来主要讨论开发 MSG 模块的初衷，以及与其他线程间通讯方式的对比。

开发的初衷是受 ROS 及其他学校的如吉林大学的软总线中心，湖南大学的消息中心模块的启发，（主要还是苦全局变量久矣），开发一套简单易用、线程安全、模块解耦、逻辑清晰的线程间通讯机制显得尤为重要。

但是 RTOS 中不是已经提供了诸如：邮箱，消息队列、信号（软中断，这里不具体讨论）的线程间通信机制，为什么不直接用呢？接下来就对各个通信方式进行对比分析：

邮箱：邮箱中的每一封邮件只能容纳固定的 4 字节内容，当需要传输更大的数据时，可以将指针作为邮件的内容进行传输。但使用邮箱有一个问题：接收邮件，就会将邮件取出，也就是说，发出去的信只有一份，被一个线程接收后，其他的线程就收不到这封信了，因此面对多个消费者的情况时就会出现问题。而且邮箱的机制是必须要现有生产者发出一封邮件，对应的消费者才有邮件可读，也就是说，每次读取其实是阻塞性质的，生产者的邮件发出之前，消费者是读不到数据的，也不能读取使用上一次的历史数据，因此只要生成者发送邮件不及时就容易出现问题。

但是在机器人的开发中，经常会面对多个消费者，以及当数据更新不及时但为了维持控制频率就先延用上一次数据的情况（当然，长时间数据未更新的情况下沿用历史数据是绝对不行的！轻则抖动，重则失控！例如遥控器、电机离线等，对于这些情况也有相应的离线检测和处理机制）。

消息队列：消息队列是邮箱的扩展，传输的数据就包括指向消息的指针和消息的长度，便于数据解析获取。但上述邮箱通讯方式的弊端依旧存在。

全局变量：全局变量读写直接快速，使用简单，但其如果在 RTOS 中滥用，其危害性就很恐怖了，我们应该尽量避免使用全局变量，具体的影响和危害就不展开讲了。

MSG模块：msg 模块主要解决的就是上述邮箱及消息队列面对多个消费者等问题，以及避免全局变量满天飞的情况。msg 模块较为轻量化，是对 ROS 通讯机制的拙略模仿，模仿其话题、发布者、订阅者之间的通讯机制。订阅者读取话题，并不会取出及改动话题的数据，不会影响到其他订阅者对话题的读取。并且订阅者读取话题时不是阻塞的，不需要发布者先更新话题，订阅者和发布者之间并没有先后顺序。而且其是线程安全的，使用信号量进行保护，不过读写速度肯定不如直接使用全局变量快，但基本可以忽略，利远大于弊。

代码实现

这是对话题、订阅者、发布者这几个重要对象的定义。话题对象包含名字（订阅和发布的话题主要就通过名字判断和联系起来）、指向数据的指针、提供安全性的信号量；发布者和订阅者包含话题名称、指向话题的指针、消息的长度。这里需要注意，话题中 msg 示例的数据格式，需要发布者和订阅者都知道才能正确解析，和 ROS 中的 msg 类似。

/**

• @brief 话题类型

/
typedef struct topic
{
char name[MSG_NAME_MAX];
void msg; // 指向msg实例的指针
rt_sem_t sem;
} topic_t;
/

@brief 订阅者类型.每个发布者拥有发布者实例,并且可以通过链表访问所有订阅了自己发布的话题的订阅者

*/
typedef struct sublisher
{
const char topic_name;
topic_t tp; // 话题的指针
uint8_t len; // 消息类型长度
} subscriber_t;
/

@brief 发布者类型.每个发布者拥有发布者实例,并且可以通过链表访问所有订阅了自己发布的话题的订阅者

*/
typedef struct publisher
{
const char *topic_name;
topic_t *tp; // 话题的指针
uint8_t len; // 消息类型长度
} publisher_t;

以下是订阅者和发布者注册的函数，其注册不需要分先后顺序：

/**

@brief 注册成为消息发布者

@param name 发布者发布的话题名称(话题)
@param len 消息类型长度,通过sizeof()获取
@return publisher_t* 返回发布者实例
*/
publisher_t *pub_register(char *name, uint8_t len){
uint8_t check_num = check_topic_name(name);
publisher_t *pub = (publisher_t *)rt_malloc(sizeof(publisher_t));
if(!check_num){ // 如果不存在重名的话题实例
topic_obj[idx] = (topic_t )rt_malloc(sizeof(topic_t));
rt_memset(topic_obj[idx], 0, sizeof(topic_t));
if(topic_obj[idx] == NULL){
LOG_E("malloc failed!");
return NULL;
}
topic_obj[idx]->msg = (void )rt_malloc(len);
if(topic_obj[idx]->msg == NULL){
LOG_E("malloc failed!");
return NULL;
}
topic_obj[idx]->sem = rt_sem_create(name, 1, RT_IPC_FLAG_PRIO);
rt_strcpy(topic_obj[idx]->name, name);
pub->tp = topic_obj[idx];
pub->topic_name = topic_obj[idx]->name;
pub->len = len;
idx++;
}
else{
pub->tp = topic_obj[check_num - 1];
pub->topic_name = topic_obj[check_num - 1]->name;
pub->len = len;
}
return pub;
}
/
@brief 订阅name的话题消息

@param name 话题名称

@param data 消息长度,通过sizeof()获取
@return subscriber_t* 返回订阅者实例
*/
subscriber_t *sub_register(char *name, uint8_t len){
// 和发布者同样的流程，直接调用发布者的注册函数
subscriber_t *sub = (subscriber_t *)pub_register(name, len);
return sub;
}

注册时通过传入的话题名称和消息长度创建对应的话题，会先检查目前已有的话题中是否有同样名称的话题，如有则返回已有的话题地址，如果没有那就创建新的话题。

因此，话题的名称很重要，如果订阅者和发布者创建时名称差了一个字符就会对不上，推荐话题的名称直接使用宏定义进行管理，避免出错 （否则出错了很难查的）

以下是订阅者和发布者处理话题的简单实现：

/**

• @brief 发布消息
• @param pub 发布者实例指针
• @param data 数据指针,将要发布的消息放到此处
• @return uint8_t 返回值为0说明发布失败,为1说明发布成功
  */
  uint8_t pub_push_msg(publisher_t pub, void data){
  if(rt_sem_take(pub->tp->sem, RT_WAITING_NO)){
  LOG_W("take sem failed!");
  return 0;
  }
  rt_memcpy(pub->tp->msg, data, pub->len);
  rt_sem_release(pub->tp->sem);
  return 1;
  }
  /
  @brief 获取消息

@param sub 订阅者实例指针
@param data 数据指针,接收的消息将会放到此处
@return uint8_t 返回值为1说明获取到了新的消息
*/
uint8_t sub_get_msg(subscriber_t *sub, void *data){
rt_memcpy(data, sub->tp->msg, sub->len);
return 1;
}

可以看出目前是十分简陋的（轻量化（不是），只有在发布者更新话题时使用信号量进行了保护，从而保证写入的完整性。但订阅者获取消息是没有保护的，因为只要保证了发布者更新数据的完整性，订阅读取话题是没有问题的。

但其也是还有许多需要完善的地方，鲁棒性仍需加强，例如 rt_memcpy 并不保证原子性。这里主要就是提供一个解决思路，抛砖引玉。

使用示例
/* 首先需要在.h文件中定义话题的数据格式 /
struct msg_test{
uint8_t id;
uint8_t data[5];
}
/ 在发布者的.c文件中 */
publisher_t pub;
stuct msg_test msg_p;
pub = pub_register("msg_test", sizeof(struct msg_test));
msg_p.id = 1;
for(uint8_t i = 0, i < 5, i++){
msg_p.data[0] = i;
}
pub_push_msg(pub, &msg_p);
/ 在订阅者的.c文件中 */
subscriber_t *sub;
stuct msg_test msg_s;
sub = sub_register("msg_test", sizeof(struct msg_test));
sub_get_msg(sub, &msg_s);