第4章 uORB深入理解

4. uORB深入理解：消息发布/订阅机制、消息定义与生成、多线程同步

uORB，全称是Micro Object Request Broker。说白了，它就是PX4里所有模块之间通信的“高速公路”。

我刚开始接触PX4时，第一反应是：为什么不用ROS？后来踩了几个坑才明白，在嵌入式实时系统里，ROS那套太重了。uORB轻量、无锁、实时性好，这才是飞控该有的样子。

4.1 消息发布/订阅机制

uORB的核心模式，就是发布-订阅。一个模块发布消息，其他模块订阅它。发布者和订阅者互不知道对方的存在。

举个例子。传感器模块发布原始IMU数据，姿态估计模块订阅它。姿态估计模块发布姿态数据，位置估计模块再订阅。每个模块只关心自己需要的消息。

核心概念：

发布者(Publisher)：谁产生数据，谁就是发布者
订阅者(Subscriber)：谁需要数据，谁就是订阅者
主题(Topic)：消息的唯一标识，比如sensor_accel
消息实例(Instance)：同一个主题可以有多个实例，比如多个IMU

我见过不少新手犯一个错误：在回调函数里直接发布消息。嗯，这里要注意，uORB的发布是同步的，如果你在中断上下文里发布，可能会造成死锁。

4.1.1 发布消息的典型流程

// 1. 先广告(advertise)这个主题
orb_advert_t pub = orb_advertise(ORB_ID(sensor_accel), &accel_data);

// 2. 然后循环发布
while (true) {
    // 填充数据
    accel_data.timestamp = hrt_absolute_time();
    accel_data.x = read_accel_x();
    accel_data.y = read_accel_y();
    accel_data.z = read_accel_z();

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

    // 控制频率
    px4_usleep(1000); // 1kHz
}

我个人习惯把orb_advertise放在模块初始化阶段。为什么？因为广告操作会创建内核对象，如果频繁广告/取消广告，会有性能开销。

4.1.2 订阅消息的典型流程

// 1. 订阅主题
int sub_fd = orb_subscribe(ORB_ID(sensor_accel));

// 2. 设置队列大小（可选）
orb_set_interval(sub_fd, 10); // 10ms内不重复接收

// 3. 循环读取
while (true) {
    struct sensor_accel_s accel_data;
    orb_copy(ORB_ID(sensor_accel), sub_fd, &accel_data);

    // 处理数据
    process_accel(&accel_data);

    // 等待新数据
    px4_pollfd_struct_t fds[] = {{ .fd = sub_fd, .events = POLLIN }};
    px4_poll(fds, 1, 1000); // 超时1秒
}

我曾经遇到过一个bug：订阅者读取数据时，发现时间戳总是旧的。排查了半天，发现是忘了调用orb_check检查是否有新数据。你想想看，orb_copy只是拷贝当前最新数据，如果没有新数据，它返回的还是旧值。

4.2 消息定义与生成

uORB的消息定义，用的是.msg文件。这些文件放在msg/目录下。编译时，代码生成器会自动生成C/C++结构体和对应的ORB_ID。

我刚开始看的时候觉得多此一举，直接手写结构体不就行了？后来发现，自动生成的好处太多了：

保证所有模块使用同一份定义
自动生成序列化/反序列化代码
支持多语言绑定（C、C++、Python）

4.2.1 消息定义文件示例

拿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  # 队列深度

注意那个ORB_QUEUE_LENGTH。我建议你根据实际需求调整这个值。如果发布频率远高于订阅频率，队列太短会丢数据；太长又浪费内存。

4.2.2 代码生成过程

编译时，msg/目录下的.msg文件会被px4_msg_gen.py脚本处理。生成的文件在build/<target>/msg/下。

生成的内容包括：

结构体定义：比如struct sensor_accel_s
ORB_ID宏：比如ORB_ID(sensor_accel)
序列化函数：sensor_accel_s::serialize()
哈希值：用于运行时校验消息版本

避坑指南：

我曾经修改了一个.msg文件，加了几个字段。结果忘记重新编译所有依赖模块，导致运行时订阅者拿到的数据全是乱的。因为消息的二进制布局变了，但订阅者还在用旧的结构体解析。

解决办法：修改.msg后，一定要make clean && make全量编译。

4.3 多线程同步

uORB本身是线程安全的。它的内部实现用了原子操作和内存屏障，没有用锁。这也是它能在实时系统中高效运行的原因。

但是，这不意味着你可以在多线程中随意使用uORB。我见过太多因为线程同步问题导致的诡异bug。

4.3.1 发布者的线程安全

同一个主题的发布者，可以在多个线程中同时发布吗？答案是可以，但我不建议这么做。

为什么？因为orb_publish内部会更新一个全局的“最新数据指针”。如果两个线程同时发布，可能会出现数据覆盖：

// 线程A
accel_data.x = 1.0;
orb_publish(ORB_ID(sensor_accel), pub, &accel_data);

// 线程B
accel_data.x = 2.0;
orb_publish(ORB_ID(sensor_accel), pub, &accel_data);

订阅者可能读到1.0，也可能读到2.0，甚至读到半更新状态（虽然uORB内部用了内存屏障，但结构体本身不是原子更新的）。

警告：

如果你必须在多线程中发布同一个主题，请确保每个线程使用独立的orb_advert_t句柄。或者，更稳妥的做法：用一个线程专门负责发布，其他线程通过队列把数据传给它。

4.3.2 订阅者的线程安全

订阅者这边相对安全。因为orb_copy是读操作，多个线程同时读同一个订阅句柄不会有问题。

但要注意：orb_check和orb_copy不是原子操作。如果你先检查到有新数据，然后去拷贝，中间可能有其他线程把数据读走了。虽然概率很低，但在高实时性要求下需要考虑。

我个人习惯的做法是：每个订阅者句柄只在一个线程中使用。如果多个线程需要同一份数据，就各自订阅一次。

4.3.3 使用poll机制避免忙等

很多新手写订阅循环时，喜欢这样：

while (true) {
    orb_copy(ORB_ID(sensor_accel), sub_fd, &data);
    // 处理数据
}

这是典型的忙等，CPU占用率直接拉满。正确的做法是用px4_poll：

px4_pollfd_struct_t fds[] = {
    { .fd = sub_fd, .events = POLLIN }
};

while (true) {
    int ret = px4_poll(fds, 1, 100); // 等待100ms

    if (ret > 0) {
        // 有数据来了
        orb_copy(ORB_ID(sensor_accel), sub_fd, &data);
        process_data(&data);
    } else if (ret == 0) {
        // 超时，可以做些其他事情
        check_health();
    } else {
        // 出错
        PX4_ERR("poll failed");
    }
}

这样CPU只在有数据时才被唤醒，其他时间可以休眠或者处理其他任务。

4.4 核心知识体系

下面这张图，是我自己总结的uORB核心逻辑。你看一遍就能明白整个数据流是怎么走的。

4.5 实战经验总结

最后，分享几个我这些年积累的uORB使用心得：

场景	建议做法	原因
高频传感器数据	使用队列深度≥4	防止订阅者处理不及时导致丢数据
低频控制指令	队列深度=1即可	只需要最新指令，旧指令无意义
多线程发布同一主题	使用独立句柄+互斥锁	避免数据竞争和覆盖
调试时查看消息	使用`uorb top`命令	实时查看各主题发布频率和延迟
消息内容变更	全量编译+重启飞控	确保所有模块使用新定义

一个小技巧：

调试uORB问题时，我经常在代码里加一句：

PX4_INFO("sensor_accel: %.2f %.2f %.2f", accel_data.x, accel_data.y, accel_data.z);

然后通过dmesg或者串口看输出。虽然粗暴，但往往能最快定位问题。

uORB的设计哲学就是“简单、可靠、实时”。你只要理解了发布/订阅模型，掌握了消息定义和生成流程，再注意多线程同步的几个坑，基本上就能驾驭它了。

嗯，这一章就到这里。记住，uORB不是ROS，别拿ROS那套思维来套它。轻量、高效、无锁，这才是飞控该有的样子。