ROS2系列教程：参数Parameter与QoS策略

1
参数名（parameter name）：max_speed
2
参数值（value）：2.5
3
参数类型（type）：DOUBLE
4
所属节点（node）：/robot_controller

1
# param_declare.py —— 声明参数
2
import rclpy
3
from rclpy.node import Node
4

5
class ParamNode(Node):
6
    def __init__(self):
7
        super().__init__('param_node')
8
        # 声明参数：名称 + 默认值 + 描述（描述需 Humble+）
9
        self.declare_parameter('robot_name', 'turtle_01', '机器人名称')
10
        self.declare_parameter('max_speed', 2.5, '最大线速度')
11
        self.declare_parameter('enable_safety', True, '是否启用安全功能')
12
        self.declare_parameter('sensor_ids', [1, 2, 3], '传感器ID列表')
13

14
        # 读取参数（返回 Parameter 对象，用 .value 取原始值）
15
        self.robot_name = self.get_parameter('robot_name').value
16
        self.max_speed = self.get_parameter('max_speed').value
17
        self.enable_safety = self.get_parameter('enable_safety').value
18
        self.sensor_ids = self.get_parameter('sensor_ids').value
19

20
        self.get_logger().info(
21
            f'机器人={self.robot_name} 速度={self.max_speed} '
22
            f'安全={self.enable_safety} 传感器={self.sensor_ids}')
23

24
def main():
25
    rclpy.init()
26
    node = ParamNode()
27
    rclpy.spin(node)
28
    node.destroy_node()
29
    rclpy.shutdown()
30

31
if __name__ == '__main__':
32
    main()

1
# param_callback.py —— 参数变更回调
2
import rclpy
3
from rclpy.node import Node
4

5
class ParamCallbackNode(Node):
6
    def __init__(self):
7
        super().__init__('param_callback_node')
8
        self.declare_parameter('max_speed', 2.5)
9
        # 注册参数回调：返回 SetParametersResult
10
        self.add_on_set_parameters_callback(self.on_param_change)
11
        self.current_speed = self.get_parameter('max_speed').value
12
        self.get_logger().info(f'初始速度: {self.current_speed}')
13

14
    def on_param_change(self, params):
15
        for param in params:
16
            if param.name == 'max_speed':
17
                if param.value > 10.0:
18
                    self.get_logger().warn(f'速度 {param.value} 过大，拒绝！')
19
                    return rclpy.parameter.SetParametersResult(successful=False)
20
                self.current_speed = param.value
21
                self.get_logger().info(f'速度已更新为: {self.current_speed}')
22
        return rclpy.parameter.SetParametersResult(successful=True)
23

24
def main():
25
    rclpy.init()
26
    node = ParamCallbackNode()
27
    rclpy.spin(node)
28
    node.destroy_node()
29
    rclpy.shutdown()
30

31
if __name__ == '__main__':
32
    main()

1
# Python 读取嵌套参数
2
p_gain = self.get_parameter('pid_gains.p').value   # 点号访问
3
i_gain = self.get_parameter('pid_gains.i').value

1
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
2
from rclpy.qos import qos_profile_sensor_data, qos_profile_system_default
3

4
# 1. 传感器数据档（BEST_EFFORT + KEEP_LAST(5)）：图像/雷达常用
5
#    丢几帧没关系，要的是低延迟
6
qos_profile_sensor_data
7

8
# 2. 默认档（RELIABLE + KEEP_LAST(10)）：控制指令等关键数据
9
#    保证送达，延迟可接受
10

11
# 3. 系统默认档：节点默认行为
12
qos_profile_system_default

1
# qos_demo.py —— 自定义 QoS
2
import rclpy
3
from rclpy.node import Node
4
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy, DurabilityPolicy
5
from std_msgs.msg import String
6

7
class QosNode(Node):
8
    def __init__(self):
9
        super().__init__('qos_node')
10
        # 自定义 QoS：可靠 + 保留最近 20 条 + 瞬态本地
11
        qos = QoSProfile(
12
            reliability=ReliabilityPolicy.RELIABLE,
13
            history=HistoryPolicy.KEEP_LAST,
14
            depth=20,
15
            durability=DurabilityPolicy.TRANSIENT_LOCAL,
16
        )
17
        self.pub = self.create_publisher(String, 'qos_topic', qos)
18
        self.sub = self.create_subscription(String, 'qos_topic', self.on_msg, qos)
19
        self.timer = self.create_timer(1.0, self.tick)
20

21
    def tick(self):
22
        msg = String(); msg.data = 'qos test'
23
        self.pub.publish(msg)
24

25
    def on_msg(self, msg):
26
        self.get_logger().info(f'收到: {msg.data}')
27

28
def main():
29
    rclpy.init()
30
    node = QosNode()
31
    rclpy.spin(node)
32
    node.destroy_node()
33
    rclpy.shutdown()
34

35
if __name__ == '__main__':
36
    main()

1
// cpp_pkg/src/qos_sensor.cpp —— 传感器 QoS 订阅
2
#include "rclcpp/rclcpp.hpp"
3
#include "sensor_msgs/msg/laser_scan.hpp"
4

5
class ScanSubscriber : public rclcpp::Node
6
{
7
public:
8
  ScanSubscriber() : Node("scan_subscriber")
9
  {
10
    // BEST_EFFORT：激光雷达丢帧可接受，要低延迟
11
    auto qos = rclcpp::QoS(rclcpp::KeepLast(5))
12
        .best_effort()
13
        .durability_volatile();
14
    sub_ = this->create_subscription<sensor_msgs::msg::LaserScan>(
15
      "scan", qos,
16
      std::bind(&ScanSubscriber::on_scan, this, std::placeholders::_1));
17
  }
18

19
private:
20
  void on_scan(const sensor_msgs::msg::LaserScan::SharedPtr msg)
21
  {
22
    RCLCPP_INFO(this->get_logger(), "收到 %zu 个扫描点", msg->ranges.size());
23
  }
24

25
  rclcpp::Subscription<sensor_msgs::msg::LaserScan>::SharedPtr sub_;
26
};
27

28
int main(int argc, char** argv)
29
{
30
  rclcpp::init(argc, argv);
31
  rclcpp::spin(std::make_shared<ScanSubscriber>());
32
  rclcpp::shutdown();
33
  return 0;
34
}

1
规则 1：Reliability 要求发布端 >= 订阅端（RELIABLE 发，BEST_EFFORT 收 OK；反向不 OK）
2
规则 2：Depth 等其余策略取双方协商值（取更宽松）
3
规则 3：Durability TRANSIENT_LOCAL 的发布者，晚加入订阅者也能收到历史消息