背景

使用的開發板為大疆的 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);