ROS2 话题通讯原理（MQ5 气体传感器）

目标本课学完，你应该能做到

01跑出现象：一步步跑出 MQ5 浓度在话题上流动

先别急着啃原理——把系统跑起来，看它到底长什么样。下面按终端分成四段，一条命令一张卡：每条都告诉你为什么敲它，以及敲完屏幕上应该出现什么。逐条对照，现象对上了；再带着「为什么会这样」的疑问进入后面的原理与代码，会轻松很多。

# ① 进工作空间，并先加载 ROS2 本体（underlay）
cd ~/ros2_ws
source /opt/ros/humble/setup.bash
# ② 只编译本实验需要的两个包：节点包 + 它依赖的接口包
#    用 --packages-select 点名，不碰工作空间里其它包，快且不易连带报错
colcon build --packages-select ros_sensor ros_sensor_interfaces
# ③ 把工作空间注册进当前终端（新终端必做，不能省）
source install/setup.bash

2运行 MQ5 采集节点终端 B

这就是数据的源头：它每秒读一次 ADC，把电压换算成 ppm 浓度（换算公式在第 4 节讲），然后把数值"喊"到 /gas_sensor 这个话题上。

# 新终端必做：进目录 + 加载本工作空间（不然找不到 ros_sensor）
cd ~/ros2_ws
source install/setup.bash
# 启动 MQ5 采集节点（普通用户即可，不需要 sudo）
ros2 run ros_sensor gas_sensor_node

$ ros2 run ros_sensor gas_sensor_node [INFO] [1728000001.123] [gas_sensor_node]: raw=1404 voltage=1.714V ppm=312.91 [INFO] [1728000002.124] [gas_sensor_node]: raw=1405 voltage=1.715V ppm=313.68 [INFO] [1728000003.126] [gas_sensor_node]: raw=1403 voltage=1.713V ppm=312.15 …每秒一行，持续滚动

✔ 每秒稳定打印一行 raw=… voltage=…V ppm=…：raw 是 ADC 原始值、voltage 是换算出的电压、ppm 是最终浓度。光标停住不返回，正是节点在持续工作的样子，停止请按 Ctrl+C。

3运行报警监控节点终端 C

它订阅同一个话题。注意：它完全不知道发布端在哪、是谁——只知道"我去 /gas_sensor 这个频道听"。

# 同样的固定开场：进目录 + source（新终端必做）
cd ~/ros2_ws
source install/setup.bash
# 启动订阅 / 报警节点
ros2 run ros_sensor gas_monitor

$ ros2 run ros_sensor gas_monitor [INFO] [gas_monitor]: 浓度正常: 312.91 ppm [INFO] [gas_monitor]: 浓度正常: 313.68 ppm [INFO] [gas_monitor]: 浓度正常: 312.15 ppm

✔ 数值与终端 B 的 ppm 一一对应——这就是消息真的走通了。默认阈值 500 ppm，所以此刻打印的是"浓度正常"。

# 新终端开场：进目录 + source（后面的命令都在这个终端里敲，不用再 source）
cd ~/ros2_ws
source install/setup.bash
# 给整个系统的节点"点名"
ros2 node list

# 列出当前活跃的话题（本终端已 source 过，直接敲）
ros2 topic list

# 看这条话题的类型 + 谁在发、谁在收
ros2 topic info /gas_sensor
# 再加一把 --verbose：还能看到两端的 QoS 参数
ros2 topic info /gas_sensor --verbose

# 变成第三个订阅者实时"偷听"；随时 Ctrl+C 停止
ros2 topic echo /gas_sensor

# 测这条话题的发布频率；随时 Ctrl+C 停止
ros2 topic hz /gas_sensor

# 在终端 B 按 Ctrl+C 停掉，再用参数重启
ros2 run ros_sensor gas_sensor_node --ros-args -p publish_period:=0.5

# 在终端 C 按 Ctrl+C 停掉，再用参数重启
ros2 run ros_sensor gas_monitor --ros-args -p alarm_threshold:=100.0

02问题引入与原理：一份数据，为什么要经过"话题"

刚才那一屏数据已经跑通了。但它凭什么能跑通？先看需求——想象你在家做一个"燃气安全管家"：厨房装了 MQ5 可燃气体传感器，一旦浓度超标要立刻做四件事：手机弹通知、客厅声光报警器响、云端留一份记录、物业大屏同步显示。传感器每秒读一次数据，这四个"想听的人"都要拿到同一份。

如果不用话题，直接让传感器程序"挨个通知"这四个人呢？这是最朴素的点对点直连，一上手就会撞到四堵墙：

话题通信把这四堵墙一次性拆掉：传感器只管把消息放到 /gas_sensor 这个话题上，谁想用谁就自己"竖起耳朵"订阅，随时加入、随时离开，发布者永远不需要知道现在有几个听众。

那它内部到底怎么跑通的？四个环节

"解耦"听起来抽象，其实全过程只有四步，每一步都有明确的承担者。看这张结构图——它把两个节点的内部拆开了，你会发现真正干活的并不是 publish() 那一行，而是它背后 DDS 的"收发管道"。

03QoS：给话题签一份传输合同，并亲手签坏它

同样是发数据，激光雷达"丢几帧无所谓、延迟必须低"，控制指令却"一条都不能丢"。ROS2 把这个选择权交给了 QoS（Quality of Service）：每个话题都附带一套传输策略——发布端提供一套，订阅端要求一套，DDS 逐项比对，匹配成功才建立连接。

按下场景选档位

动手验证：把合同"签坏"看看

光看懂规则不够，下面这个实验台把抽象的兼容规则变成了几秒就能验证的实验：左边是发布端 gas_sensor_node，右边是订阅端 gas_monitor，改完档位点一下"跑 10 条消息"，模拟器就按 DDS 的规则逐条演示。建议按这张任务单操作，边做边对照上面的规则：

🎛️ 实验台就在下面：改两端档位，点"跑 10 条消息"，按上表的三个任务依次验证。

04实验步骤：先看懂框架，再读代码，最后自己跑出结果

这一节换个顺序：先看程序骨架（三份文件各管什么、谁调谁）→ 再逐段读关键代码（只讲非讲不可的几十行）→ 最后动手跑、把结果和预期对上。反过来"先啃代码"最容易迷路——你会记住语法，却说不清这段代码为什么长这样。

第 1 步 · 先看骨架：三份文件，一条数据流

刚才屏幕上的一切，来自三份代码。别急着看细节，先记住它们的分工——这份骨架看懂了，后面读代码只是填肉。

第 2 步 · 再读代码：三段非讲不可的地方

下面按数据流动的顺序看三段代码。每段只抓关键几行——其余的 import、初始化、日志格式，你打开源码一眼就懂，这里不占篇幅。

① common.py —— ppm 是算出来的，不是读出来的

ppm 是什么在 01 节已经说过（parts per million，百万分之一）。这一节补的关键点是：它不是传感器直接读出来的数字。MQ5 是半导体气敏元件，接触可燃气体时内阻下降；电路通过分压把这个电阻变化变成电压。而"电压"和"浓度"不是线性关系——它的灵敏特性曲线只有在双对数坐标下才近似直线，所以必须在对数域里插值，才能把电压反推成 ppm。

# 来源：ros_sensor/common.py L13-33
V_REF, RL, R0 = 5.0, 1.0, 10.0      # 参考电压、负载电阻 kΩ、标定电阻 kΩ
CURVE_B, CURVE_M = 0.186, -0.362  # 灵敏特性曲线截距与斜率（斜率为负！）

def voltage_to_ppm(v):                        # 电压 → 浓度
    if v <= 0 or v >= V_REF:                 # ← 除零保护：量程外的电压不算
        return 0.0
    rs    = RL * (V_REF - v) / v               # ③ 由分压关系反推气敏电阻 Rs
    ratio = rs / R0                            # ③ 与标定点 R0 的比值
    return 10 ** ((math.log10(ratio) - CURVE_B) / CURVE_M)   # ④ 对数域线性插值

采集入口：read_mq5_once() 每秒钟被调用一次

# 来源：ros_sensor/common.py L36-47
def read_mq5_once():
    try:
        scale = float(read_adc_attr("in_voltage_scale"))      # ① 量程系数：1 格 = 多少毫伏
        raw   = int(read_adc_attr(f"in_voltage{MQ5_CH}_raw"))   # ② 原始格数（12 bit，0~4095）
        v = raw * scale / 1000.0                          # ③ 格数×毫伏/格=毫伏，÷1000→伏特
        return raw, v, voltage_to_ppm(v), ""            # ④ 一次返回四件事
    except Exception as e:
        return None, None, None, f"{type(e).__name__}: {e}"   # 失败信息交给调用方

四个返回值 (raw, v, ppm, err) 就是节点日志里那一行的全部来源。函数不打印、不抛异常，只把结果和错误一起交出去——打印交给节点，这是数据层与表现层分离的好习惯。

② gas_sensor_node.py —— 发布端只做四件事

# 来源：ros_sensor/gas_sensor_node.py L16-44
from rclpy.node import Node
from std_msgs.msg import Float32
from rclpy.qos import QoSProfile, HistoryPolicy, ReliabilityPolicy
from .common import read_mq5_once                # 复用 ① 里的采集函数

class GasSensorNode(Node):
    def __init__(self):
        super().__init__('gas_sensor_node')                       # ① 注册节点名
        self.declare_parameter('publish_period', 1.0)         # ② 声明可调参数

        qos = QoSProfile(                                          # ③ 传输合同
            history=HistoryPolicy.KEEP_LAST, depth=10,
            reliability=ReliabilityPolicy.RELIABLE,
        )
        self.pub = self.create_publisher(Float32, '/gas_sensor', qos)   # ④ 创建发布端

        period = float(self.get_parameter('publish_period').value)
        self.timer = self.create_timer(period, self.timer_cb)      # ⑤ 定时器：每 period 秒叫一次

    def timer_cb(self):                                            # ⑥ 每秒被触发一次的回调
        raw, v, ppm, err = read_mq5_once()                          #    读 + 算
        if err:
            self.get_logger().error(f'读取 ADC 失败: {err}')       #    出错只记日志，不崩溃
            return
        msg = Float32()                                             #    装消息
        msg.data = float(ppm)
        self.pub.publish(msg)                                       # ⑦ 发布：入队即返回
        self.get_logger().info(f'raw={raw} voltage={v:.3f}V ppm={ppm:.2f}')

def main():
    rclpy.init()                    # 初始化 rclpy
    node = GasSensorNode()          # 构造节点（此时才注册进计算图）
    try: rclpy.spin(node)         # 进入事件循环：驱动定时器与回调
    except KeyboardInterrupt: pass
    finally: node.destroy_node(); rclpy.shutdown()   # Ctrl+C 后干净退出

③ gas_monitor.py —— 订阅端与发布端镜面对称

# 来源：ros_sensor/gas_monitor.py L13-36
class GasMonitor(Node):
    def __init__(self):
        super().__init__('gas_monitor')                        # ① 另一个节点名
        self.declare_parameter('alarm_threshold', 500.0)    # ② 报警阈值也是参数
        self.threshold = float(self.get_parameter('alarm_threshold').value)

        qos = QoSProfile(                                       # ③ QoS 必须与发布端相容
            history=HistoryPolicy.KEEP_LAST, depth=10,
            reliability=ReliabilityPolicy.RELIABLE,
        )
        self.sub = self.create_subscription(                    # ④ 订阅：类型 + 话题 + 回调 + QoS
            Float32, '/gas_sensor', self.callback, qos)

    def callback(self, msg):                                    # ⑤ 消息到了才会被调用
        ppm = msg.data
        if ppm >= self.threshold:
            self.get_logger().warn(f'[报警] 气体浓度超标: {ppm:.2f} ppm >= {self.threshold} ppm')
        else:
            self.get_logger().info(f'浓度正常: {ppm:.2f} ppm')

第 3 步 · 最后跑出结果：把代码和现象对上号

代码读完了，现在回到终端，验证你读到的每一段在屏幕上长什么样。这一步不是重复 01 节，而是让"代码里的某一行"和"屏幕上的某一行"建立一一对应。

05动手任务：改点新东西，验证你真的懂了

01 节和 04 节第 3 步都是"跟着看现象"。这一节的任务故意不重复那些操作——每个任务都要你改一处新东西（改代码、改参数、加节点），然后自己判断"结果应该是什么样"再去验证。能预判结果，才算真懂。

06注意事项与验收清单

• 双节点运行，ros2 node list 同时显示两个节点
• ros2 topic info /gas_sensor 显示 Pub=1、Sub≥1，类型 std_msgs/msg/Float32
• ros2 topic hz ≈ 1.000 Hz，改 publish_period:=0.5 后变 ≈ 2Hz
• 第三方 ros2 topic echo 可同时接收（一对多验证）
• 调低 alarm_threshold:=100.0 后出现 [报警] 日志
• 口头解释：为什么发布端不用知道订阅端 IP？MQ5 对数换算四步链路？

07实验总结与思考题