模块之间为什么要通信？#

上位机不是一个 main 搞定所有事。串口驱动要收发硬件数据，决策状态机要发指令等反馈，传感器处理要跑雷达和视觉——这些东西同时在跑，还得互相传数据。串口驱动收到轮速得给决策用，决策发出速度指令得给串口驱动发出去。

最粗暴的做法是直接函数调用，但模块跑在不同线程甚至不同进程里，直接调不是锁死就是压根调不到。所以得有个中间人帮忙转发，这东西就叫消息总线。

ROS2 Pub/Sub#

2026 年做 RC 上位机的人大概率绕不开 ROS2。它继承了 ROS 1 十几年的生态，DDS 通信、节点管理、参数系统一应俱全，行业里的 SLAM、导航、视觉方案几乎都挂在上面。可以说 ROS2 就是机器人软件的事实标准。

但 RC 赛场不是工厂车间。一辆竞速赛车的上位机可能只需要串口驱动、一个决策状态机、一套 Pure Pursuit——总共三四个节点，跑在一台 Jetson 上，通信频率不过 50Hz。为这几个节点装一整套 ROS2 + DDS，就像为了喝杯水去建自来水厂。能喝到水吗？能。值不值是另一回事。

先不管值不值，ROS2 的 pub/sub 得会用，因为很多现成的东西（串口驱动、雷达驱动、定位算法）都挂在 ROS2 上。

1
import rclpy
2
from rclpy.node import Node
3
from std_msgs.msg import String
4

5
class Talker(Node):
6
    def __init__(self):
7
        super().__init__('talker')
8
        self.pub = self.create_publisher(String, '/chatter', 10)
9
        self.timer = self.create_timer(1.0, self.tick)
10

11
    def tick(self):
12
        msg = String()
13
        msg.data = 'hello'
14
        self.pub.publish(msg)
15

16
rclpy.init()
17
rclpy.spin(Talker())

1
import rclpy
2
from rclpy.node import Node
3
from std_msgs.msg import String
4

5
class Listener(Node):
6
    def __init__(self):
7
        super().__init__('listener')
8
        self.create_subscription(String, '/chatter', self.on_msg, 10)
9

10
    def on_msg(self, msg):
11
        self.get_logger().info(f'Received: {msg.data}')
12

13
rclpy.init()
14
rclpy.spin(Listener())

1
float32 x
2
float32 y
3
float32 yaw
4
int32 stair
5
int32 block
6
int32 spearhead
7
int32 area

1
┌─────────────────┐     /command      ┌──────────────────┐
2
│  C++ 串口驱动    │ ←───────────────  │  Python 决策节点   │
3
│                 │  /decision       │  async/await 决策 │
4
│                 │ ─────────────→    │                  │
5
└─────────────────┘                   └──────────────────┘

1
# 订阅 /decision，收到后触发事件唤醒协程
2
self.create_subscription(Ack, '/decision', self.act.on_upper_ack, qos)
3

4
def on_upper_ack(self, msg):
5
    if msg.up_free == 2:
6
        self.post_event(Event("ARM_DONE", success=True))

ROS2 在 RC 赛场上的问题#

用归用，不满意的地方也得说。

启动慢。 ROS2 底层是 DDS，启动时要节点发现、Topic 匹配、QoS 协商。光 rclpy.init() 这一行就要几百毫秒到几秒。赛场上按完启动按钮车 2 秒后才动，这 2 秒可能就是 DDS 在握手。

进程内通信也要序列化。 不管发布者和订阅者是不是在同一个进程里，消息都要走一遍序列化 → 传输 → 反序列化。一个 {x: 1.0, y: 2.0} 变成字节再变回来，在 50Hz 控制频率下这个开销不是理论上的，是能感知到的。

spin() 单线程，回调互相卡。 rclpy.spin() 默认单线程跑所有回调。视觉处理耗时 200ms 的话，这 200ms 内急停回调也收不到。我后来把决策逻辑放到单独的 asyncio 线程里才解决，但这本身就说明 ROS2 默认模型不适合实时决策——你得绕过它的限制才能用好。

装起来重。 好几个 G，编译自定义消息要配 CMakeLists.txt、package.xml、setup.py。RC 赛车就那么几个 Topic，杀鸡用牛刀。

自研 EventBus#

很多人（包括我之前）都想：ROS2 太重了，我自己写个 pub/sub 不就行了？

核心逻辑确实不复杂。C++ 版用模板和 std::any 实现类型擦除，让一个 EventBus 能传任意类型的数据，不用像 ROS2 那样提前定义 .msg 文件：

1
class EventBus {
2
    // type_index → [回调列表]，用类型索引区分不同消息
3
    std::unordered_map<std::type_index,
4
        std::vector<std::function<void(const std::any&)>>> subs_;
5
    std::mutex mtx_;
6

7
public:
8
    template<typename T>
9
    void subscribe(const std::string& topic, std::function<void(const T&)> cb) {
10
        std::lock_guard<std::mutex> lock(mtx_);
11
        subs_[std::type_index(typeid(T))].push_back(
12
            [cb](const std::any& data) { cb(std::any_cast<T>(data)); });
13
    }
14

15
    template<typename T>
16
    void publish(const std::string& topic, const T& data) {
17
        std::lock_guard<std::mutex> lock(mtx_);
18
        for (auto& cb : subs_[std::type_index(typeid(T))])
19
            cb(data);
20
    }
21
};

std::any 能装任意类型的值，std::type_index 给每个类型一个唯一的 key，这样同一个 EventBus 实例既能传 WheelSpeed 又能传 Command，编译期就做类型检查，传错类型直接抛异常。

用起来就两行：

1
EventBus bus;
2
bus.subscribe<WheelSpeed>("/wheel_speed", [](const auto& msg) {
3
    std::cout << msg.left << ", " << msg.right << std::endl;
4
});
5
bus.publish("/wheel_speed", WheelSpeed{1.0, 1.2});

Python 版更简单，一个字典加一个回调列表：

1
class EventBus:
2
    def __init__(self):
3
        self._subs = {}
4

5
    def subscribe(self, topic, callback):
6
        self._subs.setdefault(topic, []).append(callback)
7

8
    def publish(self, topic, data=None):
9
        for cb in self._subs.get(topic, []):
10
            cb(data)

但实际用起来有几个坑：

C++ 和 Python 不共享内存。 我的系统是两个进程，EventBus 在一个进程里创建，另一个进程根本访问不到。进程间通信还是得靠 ROS2、ZeroMQ 或者自己写 socket。ROS2 虽然重，但它帮你把这些都封装好了。

多线程要加锁。 串口驱动在自己的线程里 publish，决策在 asyncio 线程里 subscribe，不加锁的话回调列表会被踩坏。上面 C++ 版用了 std::mutex，但锁的粒度要控制好——锁太大了 publish 和 subscribe 互相等，锁太小了保护不住。

背压策略要自己想。 发布者 100Hz 往 Topic 扔数据，订阅者处理一帧要 30ms，队列满了怎么办？ROS2 的 QoS 帮你处理了这些，自己写的话每种策略都要自己实现。

三层架构#

不管用什么通信方式，上位机的分层是一样的。26 赛季踩了不少坑才搞清楚这件事。

1
决策调度层    状态机 / 任务规划 / 路线选择
2
  ↑ 只管"做什么"
3
控制跟踪层    Pure Pursuit / 运动学解算 / PID
4
  ↑ 只管"怎么走"
5
感知驱动层    串口驱动 / 雷达 / 视觉 / DT35
6
  只管"提供干净数据"

就和我们常说的“高内聚，低耦合”一样，各层之间通过消息总线传数据，不直接调用。感知层不知道决策层在干什么，决策层不知道串口协议长什么样。

为什么要分这么清楚？因为需求变的频率不一样。感知层的协议定了基本不动，除非换硬件；控制层调完参数基本不动，除非换底盘；但决策层——比赛前一天要改流程的话，你就得改。如果决策代码里混着串口收发逻辑，改流程的时候一不小心把协议改了，车直接寄。

拿”走到 1 号点然后抓块”这个流程举个例：

1
async def zone1(fsm, act, cfg, state):
2
    await fsm.nav_to(1.0, 2.0)        # 走
3
    await fsm.spearhead_and_wait(1)    # 抓
4
    await fsm.rotate_to(0.0, 0.7, π)  # 转

决策层不知道 Pure Pursuit 怎么算的，不知道串口发了什么字节，它只管 nav_to → 等事件 → 下一步。底下发生了什么是感知层和控制层的事。

小结#

ROS2 做 pub/sub 能用，但启动慢、序列化开销、spin 单线程这些问题是实际存在的。自研 EventBus 理论上简单，实际有跨进程、线程安全、背压等坑，全 Python 单进程可以搞，双进程老实用 ROS2 或 ZeroMQ。

三层架构的核心价值不是”代码好看”，是把变化频率不同的东西隔开——改决策不动控制，改控制不动感知。比赛前一天改方案的时候你会感谢这个分层。

下一章讲感知层——定位和视觉，搞清楚数据从哪来、什么质量。