📚 课程目录第4章参数服务器共享配置

4. 参数服务器共享配置

任务说明

在ROS2多机器人系统中，每个机器人节点需要访问统一的地图名称、速度限制等全局配置。本案例通过参数服务器（ROS2中由rclcpp::Node的declare_parameter/get_parameter机制实现）在多个机器人节点间共享配置，并演示运行时动态更新参数后所有节点同步生效的机制。

实现方案

创建两个独立节点：config_publisher负责声明并周期性更新全局参数（模拟主控端下发配置），robot_reader作为机器人端读取参数并打印当前生效配置。两个节点通过同一个参数服务器（即ROS2参数服务）交互，无需显式通信连接。

完整代码

# config_publisher.py
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import Parameter, ParameterValue, ParameterType
from rcl_interfaces.srv import SetParameters

class ConfigPublisher(Node):
    def __init__(self):
        super().__init__('config_publisher')
        # 声明全局参数（默认值）
        self.declare_parameter('map_name', 'warehouse_map')
        self.declare_parameter('max_speed', 1.5)
        self.declare_parameter('robot_count', 3)
        self.timer = self.create_timer(5.0, self.update_config)
        self.get_logger().info('配置发布节点已启动，每5秒更新一次参数')

    def update_config(self):
        # 模拟动态更新：交替切换地图与速度
        current_map = self.get_parameter('map_name').value
        new_map = 'outdoor_map' if current_map == 'warehouse_map' else 'warehouse_map'
        new_speed = 2.0 if self.get_parameter('max_speed').value == 1.5 else 1.5
        
        # 通过参数服务更新（本节点声明参数后，set_parameters可直接生效）
        self.set_parameters([
            Parameter('map_name', ParameterValue(type=ParameterType.PARAMETER_STRING, string_value=new_map)),
            Parameter('max_speed', ParameterValue(type=ParameterType.PARAMETER_DOUBLE, double_value=new_speed))
        ])
        self.get_logger().info(f'参数已更新: map={new_map}, speed={new_speed}')

def main(args=None):
    rclpy.init(args=args)
    node = ConfigPublisher()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

# robot_reader.py
import rclpy
from rclpy.node import Node

class RobotReader(Node):
    def __init__(self, robot_id):
        super().__init__(f'robot_{robot_id}_reader')
        self.robot_id = robot_id
        # 从参数服务器获取全局配置（若不存在则使用默认值）
        self.declare_parameter('map_name', 'unknown')
        self.declare_parameter('max_speed', 0.0)
        self.timer = self.create_timer(2.0, self.read_and_report)
        self.get_logger().info(f'机器人{robot_id}读取节点已启动')

    def read_and_report(self):
        map_name = self.get_parameter('map_name').value
        max_speed = self.get_parameter('max_speed').value
        self.get_logger().info(f'[机器人{self.robot_id}] 当前地图: {map_name}, 限速: {max_speed} m/s')

def main(args=None):
    rclpy.init(args=args)
    # 创建两个机器人读取节点，模拟多机器人场景
    nodes = [RobotReader(1), RobotReader(2)]
    executor = rclpy.executors.MultiThreadedExecutor()
    for n in nodes:
        executor.add_node(n)
    try:
        executor.spin()
    finally:
        for n in nodes:
            n.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

运行与验证

环境要求：ROS2 Humble或更高版本，Python3。

启动步骤：

# 终端1：启动配置发布节点
python3 config_publisher.py

# 终端2：启动两个机器人读取节点
python3 robot_reader.py

验证方法：

观察终端2中两个机器人节点每2秒打印一次配置信息，初始应显示map=warehouse_map, speed=1.5。
等待约5秒后，终端1触发参数更新，终端2中两个机器人节点在下一轮打印中应同步显示map=outdoor_map, speed=2.0。
再等待5秒，参数应再次切换回初始值，证明动态更新持续生效。

通过条件：两个机器人节点始终打印相同的参数值，且参数值随配置发布节点的更新而同步变化，无需重启任何节点。

关键说明

本案例中config_publisher通过set_parameters更新的是自身声明的参数，ROS2参数机制会自动将变更同步到参数服务器，其他节点通过get_parameter读取时即获取最新值。实际工程中若需跨节点设置参数，可使用rcl_interfaces/srv/SetParameters服务调用目标节点的参数接口，但本案例演示的是参数服务器作为全局配置中心的共享读取模式，已覆盖多机器人场景的核心需求。