uORB消息机制：PX4的"神经网络"

做嵌入式开发的人都知道，模块间通信是个绕不开的坎。在PX4里，uORB就是解决这个问题的核心机制。说白了，它就像飞控的"神经网络"——各个模块通过它来传递传感器数据、控制指令、状态信息。

我第一次接触uORB时，觉得它不就是个消息队列嘛。后来深入源码才发现，它的设计远比我想象的精巧。今天我就把uORB的底裤扒开，让你看个明白。

uORB到底是什么？

uORB的全称是micro Object Request Broker，微对象请求代理。名字挺唬人，其实核心就三件事：

• 定义消息结构——用IDL文件描述数据格式
• 发布消息——谁产生数据谁发布
• 订阅消息——谁需要数据谁订阅

这种发布-订阅模式的好处很明显：发布者和订阅者完全解耦。传感器模块不需要知道谁在用它的数据，控制模块也不需要关心数据从哪来。

消息的生命周期

一条uORB消息从诞生到消亡，经历了哪些阶段？我画了张图帮你理清思路。

消息定义：.msg文件

在PX4里，所有消息都用.msg文件定义。这些文件放在msg/目录下。我随便打开一个看看：

# sensor_accel.msg
uint64 timestamp        # 时间戳，微秒
uint64 timestamp_sample # 采样时间戳
uint32 device_id        # 设备ID
float32 x               # X轴加速度，m/s^2
float32 y               # Y轴加速度，m/s^2
float32 z               # Z轴加速度，m/s^2
float32 temperature     # 温度，摄氏度
uint8 ORB_QUEUE_LENGTH = 4  # 队列深度

字段类型支持：

• 基础类型：uint8/16/32/64、int8/16/32/64、float32/64、bool
• 数组：float32[3]、uint8[4]
• 嵌套消息：通过#include引用其他.msg
• 常量：uint8 ORB_QUEUE_LENGTH = 4

消息发布：三步走

发布一条消息，其实就三步。我拿传感器数据发布举个例子：

// 1. 声明消息结构体
struct sensor_accel_s accel_data;

// 2. 填充数据
accel_data.timestamp = hrt_absolute_time();
accel_data.x = accel_x;
accel_data.y = accel_y;
accel_data.z = accel_z;
accel_data.temperature = temp;

// 3. 发布
orb_publish(ORB_ID(sensor_accel), &accel_data);

这里有个细节：ORB_ID()宏会把消息名转换成整数ID。发布者只需要调用orb_publish()，剩下的uORB内核帮你搞定。

消息订阅：两种姿势

订阅消息有两种方式，看你的需求：

方式	函数	特点	适用场景
普通订阅	orb_subscribe()	只接收最新数据	控制模块、状态估计
队列订阅	orb_subscribe_multi()	可接收历史数据	日志记录、数据分析

订阅后怎么拿数据？标准流程是这样的：

// 1. 订阅
int sensor_sub = orb_subscribe(ORB_ID(sensor_accel));

// 2. 设置队列深度（可选）
orb_set_interval(sensor_sub, 10); // 10ms间隔

// 3. 主循环里检查并拷贝
struct sensor_accel_s accel_data;
bool updated;

while (1) {
    // 检查是否有新数据
    orb_check(sensor_sub, &updated);
    
    if (updated) {
        // 拷贝数据
        orb_copy(ORB_ID(sensor_accel), sensor_sub, &accel_data);
        
        // 处理数据
        process_accel_data(&accel_data);
    }
    
    // 其他任务...
}

为什么要先orb_check()再orb_copy()？直接拷贝不行吗？

嗯，这里有个性能考量。如果数据没更新，orb_copy()还是会做一次内存拷贝，浪费CPU。先检查再拷贝，能省掉不必要的开销。尤其是在高频率循环里，这个优化很关键。

自定义uORB消息：实战演练

假设我们要定义一个"自定义传感器融合结果"的消息。怎么做？

第一步：创建.msg文件

在msg/目录下新建my_fusion_result.msg：

# my_fusion_result.msg
uint64 timestamp
uint8 SOURCE_GPS = 1
uint8 SOURCE_VISION = 2
uint8 SOURCE_RADAR = 3

uint8 source          # 数据来源
float32 pos_x         # X位置，米
float32 pos_y         # Y位置，米
float32 pos_z         # Z位置，米
float32 vel_x         # X速度，m/s
float32 vel_y         # Y速度，m/s
float32 vel_z         # Z速度，m/s
float32 confidence    # 置信度，0-1
uint8[4] reserved     # 保留字段

第二步：编译生成头文件

运行make，PX4构建系统会自动生成my_fusion_result.h。你不需要手动写任何代码生成脚本。

第三步：在代码中使用

// 发布端
#include <uORB/topics/my_fusion_result.h>

struct my_fusion_result_s fusion;
fusion.timestamp = hrt_absolute_time();
fusion.source = my_fusion_result_s::SOURCE_VISION;
fusion.pos_x = vision_x;
fusion.pos_y = vision_y;
fusion.pos_z = vision_z;
fusion.confidence = 0.95;

orb_publish(ORB_ID(my_fusion_result), &fusion);

// 订阅端
int fusion_sub = orb_subscribe(ORB_ID(my_fusion_result));
struct my_fusion_result_s fusion_data;

while (1) {
    bool updated;
    orb_check(fusion_sub, &updated);
    
    if (updated) {
        orb_copy(ORB_ID(my_fusion_result), fusion_sub, &fusion_data);
        
        // 根据来源做不同处理
        switch (fusion_data.source) {
            case my_fusion_result_s::SOURCE_GPS:
                handle_gps_fusion(&fusion_data);
                break;
            case my_fusion_result_s::SOURCE_VISION:
                handle_vision_fusion(&fusion_data);
                break;
        }
    }
}

uORB的底层实现

你可能好奇，uORB内部到底怎么工作的？我简单说说核心机制。

uORB使用共享内存 + 发布-订阅表实现。每个消息类型对应一个发布-订阅节点，节点里维护着：

• 数据缓冲区：存储最新消息数据
• 订阅者列表：所有订阅了该消息的进程
• 更新计数器：每次发布递增，订阅者通过对比计数器判断是否有新数据

发布时，uORB把数据写入共享内存，然后通知所有订阅者。订阅者通过orb_check()检查更新计数器，有变化就调用orb_copy()把数据拷贝到自己的内存空间。

这种设计的好处是：

• 发布者和订阅者不在同一进程也能通信
• 多个订阅者共享一份数据，节省内存
• 订阅者只读，不会影响发布者

总结一下

uORB是PX4的通信基石。理解它，你就能看懂飞控里数据是怎么流动的。从消息定义、发布、订阅到自定义消息，这套机制贯穿整个PX4系统。

我个人觉得，uORB最妙的地方在于它的简洁性。没有复杂的RPC，没有繁琐的序列化，就是简单的发布-订阅。但正是这种简单，让PX4的模块化设计成为可能。

下次你写PX4模块时，记得先想清楚：你的数据从哪来，到哪去，用uORB怎么串起来。想明白了，代码自然就清晰了。