第3章 uORB消息发布与订阅

3. uORB消息发布与订阅：消息定义、发布者/订阅者模式、消息频率控制

好，咱们今天聊聊uORB。这是PX4里最核心的通信机制，说白了就是各个模块之间怎么“说话”的。你想想看，飞控里有几十个模块在同时跑——传感器、姿态估计、位置控制、电机输出……它们之间怎么交换数据？总不能每个模块都自己写一套通信协议吧？uORB就是来解决这个问题的。

我个人习惯把uORB理解成一个“消息黑板报”。谁想发布消息，就往黑板上写；谁想读消息，就来黑板上看。发布者和订阅者之间完全解耦，不需要知道对方是谁。这种设计在嵌入式系统里特别实用，尤其是我们做无人机开发的时候，模块多、迭代快，解耦能省下大把调试时间。

3.1 消息定义：先定好“黑板报”的格式

在uORB里，消息是用一个叫.msg的文件来定义的。这些文件放在msg/目录下，编译的时候会自动生成对应的C/C++结构体。我刚开始接触PX4的时候，觉得这步有点多余——不就是定义个结构体嘛，直接写头文件不就行了？后来才发现，用.msg文件的好处是：自动生成序列化/反序列化代码，还能做兼容性检查。

来看一个实际例子。这是sensor_accel.msg的简化版：

# sensor_accel.msg
uint64 timestamp          # 时间戳，单位微秒
uint64 timestamp_sample   # 采样时间戳

float32[3] x              # X轴加速度，单位m/s^2
float32[3] y              # Y轴加速度
float32[3] z              # Z轴加速度

float32 temperature       # 温度，单位摄氏度
uint32 error_count        # 错误计数

# 设备信息
uint8 ORB_QUEUE_LENGTH = 4

这里有几个关键点：

字段类型：支持uint8、int32、float32、bool等基础类型，也支持定长数组float32[3]。
时间戳：几乎所有消息都带timestamp字段，这是uORB的约定。我见过有人忘记加时间戳，结果订阅方拿到数据后不知道数据是什么时候产生的，调试起来特别痛苦。
队列长度：ORB_QUEUE_LENGTH定义了消息队列的深度。默认是1，意味着新消息会覆盖旧消息。如果你需要缓存历史数据，可以设大一点。

小技巧：定义消息时，尽量把固定长度的数组放在前面，变长数据放后面。这样内存布局更紧凑，缓存命中率也更高。这是我做性能优化时踩过的坑。

3.2 发布者/订阅者模式：谁写谁读？

消息定义好了，接下来就是怎么发布和订阅。uORB的API设计得很简洁，我总结下来就三个核心函数：

orb_advertise()：声明你要发布某类消息，返回一个发布句柄。
orb_publish()：往消息队列里写数据。
orb_subscribe()：订阅某类消息，返回一个订阅句柄。
orb_copy()：从订阅句柄里读取最新数据。

来看一个发布者的示例。假设我们要发布加速度计数据：

// 发布者代码
#include <uORB/uORB.h>
#include <uORB/topics/sensor_accel.h>

static orb_advert_t accel_pub = nullptr;

void accel_publisher_init()
{
    struct sensor_accel_s accel_data = {};
    accel_pub = orb_advertise(ORB_ID(sensor_accel), &accel_data);
    if (accel_pub == nullptr) {
        PX4_ERR("Failed to advertise accel topic");
    }
}

void accel_publisher_update()
{
    struct sensor_accel_s accel_data = {};
    accel_data.timestamp = hrt_absolute_time();
    accel_data.x[0] = read_accel_x();
    accel_data.y[0] = read_accel_y();
    accel_data.z[0] = read_accel_z();
    accel_data.temperature = read_temp();

    orb_publish(ORB_ID(sensor_accel), accel_pub, &accel_data);
}

订阅者这边更简单：

// 订阅者代码
#include <uORB/uORB.h>
#include <uORB/topics/sensor_accel.h>

int accel_sub_fd = -1;
struct sensor_accel_s accel_data;

void accel_subscriber_init()
{
    accel_sub_fd = orb_subscribe(ORB_ID(sensor_accel));
    // 设置队列大小为1，只关心最新数据
    orb_set_interval(accel_sub_fd, 0);
}

void accel_subscriber_update()
{
    bool updated = false;
    orb_check(accel_sub_fd, &updated);
    if (updated) {
        orb_copy(ORB_ID(sensor_accel), accel_sub_fd, &accel_data);
        // 处理数据...
    }
}

这里有个细节：orb_check()只是检查是否有新数据，不会阻塞。如果你需要阻塞等待，可以用orb_poll()。我个人建议在实时性要求高的循环里用orb_check()，在主循环里用orb_poll()。

注意：发布者和订阅者必须在同一个进程（或线程）内使用。uORB不支持跨进程通信。如果你需要跨进程，得用其他机制，比如共享内存或者SocketCAN。我曾经在调试一个外接传感器时，误把发布者放在一个独立进程里，结果订阅者死活收不到数据，查了半天才发现这个限制。

3.3 消息频率控制：别让数据“淹死”系统

消息频率控制是uORB里最容易出问题的地方。你想想看，如果传感器以1kHz的频率发布数据，而姿态估计模块只需要100Hz，那中间就有90%的数据是浪费的。更糟糕的是，如果订阅者处理不过来，消息队列会被填满，新数据覆盖旧数据，导致丢帧。

uORB提供了两种频率控制手段：

方法	API	适用场景
发布端限流	在`orb_publish()`前加时间判断	发布者自己控制频率，比如传感器驱动
订阅端限流	`orb_set_interval()`	订阅者指定最小更新间隔

发布端限流的代码示例：

// 发布端限流：每10ms发布一次
static hrt_abstime last_pub = 0;
constexpr hrt_abstime PUB_INTERVAL = 10000_us; // 10ms

void accel_publisher_throttled()
{
    hrt_abstime now = hrt_absolute_time();
    if (now - last_pub < PUB_INTERVAL) {
        return; // 还没到发布时间
    }
    last_pub = now;

    // 正常发布...
    orb_publish(ORB_ID(sensor_accel), accel_pub, &accel_data);
}

订阅端限流更简单：

// 订阅端限流：每20ms检查一次
orb_set_interval(accel_sub_fd, 20); // 单位毫秒

这两种方式有什么区别？我个人的经验是：

发布端限流适合传感器驱动这类“数据生产者”。比如IMU以1kHz采样，但姿态估计只需要200Hz，那驱动层直接降频发布，能减少uORB内部的数据拷贝开销。
订阅端限流适合“数据消费者”。比如日志记录模块，它可能只需要10Hz的数据，但不想修改传感器驱动的代码。这时候用orb_set_interval()最方便。

  避坑指南：我曾经在一个项目中，同时用了发布端限流和订阅端限流，结果两个频率不匹配，导致订阅者偶尔会漏掉数据。后来我统一用订阅端限流，发布端保持最高频率，问题就解决了。记住：频率控制尽量只在一端做，不要两头都设。

3.4 消息频率的调试技巧

调试消息频率时，我常用两个工具：

uorb top：在NuttShell里输入这个命令，能看到所有uORB主题的发布频率、订阅者数量、队列使用情况。比如：

nsh> uorb top
Topic                          #SUB #PUB  Freq(Hz)  Queue
sensor_accel                   3    1     1000      4
sensor_gyro                   3    1     1000      4
vehicle_attitude              2    1     200       1
vehicle_local_position        1    1     100       1

看到Freq列了吗？如果某个主题的频率远低于预期，说明发布者可能被阻塞了，或者订阅者处理不过来。

手动打时间戳：在发布者和订阅者两端分别记录时间戳，然后计算延迟。我一般会在orb_publish()前后各打一个时间戳，在orb_copy()前后也打一个，这样能精确知道数据从发布到订阅花了多少时间。

// 调试用时间戳
hrt_abstime t1 = hrt_absolute_time();
orb_publish(ORB_ID(sensor_accel), accel_pub, &accel_data);
hrt_abstime t2 = hrt_absolute_time();
PX4_INFO("Publish took %lu us", t2 - t1);

经验之谈：如果发现消息频率抖动很大，先检查发布者的调度优先级。在PX4里，传感器驱动通常跑在最高优先级，但如果你把发布者放在一个低优先级的work queue里，频率就会不稳定。我遇到过最离谱的一次，是因为某个模块里用了printf()，导致发布者被IO阻塞了200ms。

3.5 消息定义的最佳实践

最后，分享几个我在项目中总结出来的消息定义原则：

每个消息只包含一个“主题”。比如加速度计和陀螺仪分开定义，不要合成一个sensor_imu。这样订阅者可以只订阅自己需要的数据，减少内存拷贝。
时间戳用uint64，单位微秒。这是PX4的惯例，别自己发明一个float类型的时间戳，否则和其他模块对接时会很痛苦。
队列长度按需设置。默认1就够了，除非你需要历史数据。我见过有人把队列设成100，结果内存占用暴涨，而实际只用了最新的1条。
消息版本号：如果消息结构体变了，记得更新.msg文件里的版本号。uORB不会自动检测结构体变化，新旧版本混用会导致数据错乱。

嗯，关于uORB的消息定义、发布订阅和频率控制，核心内容就这些。你想想看，其实uORB的设计哲学很简单：让模块之间“松耦合”，各自管好自己的事。只要消息定义清晰、频率控制得当，整个系统就能稳定运行。我在实际调试中，80%的通信问题都出在频率不匹配或者消息定义不一致上——所以这块基础打牢了，后面会省很多事。

📚 课程目录

3. uORB消息发布与订阅：消息定义、发布者/订阅者模式、消息频率控制

3.1 消息定义：先定好“黑板报”的格式

3.2 发布者/订阅者模式：谁写谁读？

3.3 消息频率控制：别让数据“淹死”系统

3.4 消息频率的调试技巧

3.5 消息定义的最佳实践