ROS1到ROS2迁移完全指南：核心差异与迁移实践

jthou

1035人浏览 · 2025-11-21 10:24:28

jthou · 2025-11-21 10:24:28 发布

ROS1与ROS2：本质差异与联系

设计哲学的变化

ROS1 (Robot Operating System)

定位: 面向研究和原型开发
设计目标: 快速开发、易于使用
通信机制: 基于TCPROS的自定义协议
节点发现: 通过ROS Master集中式管理
实时性: 非实时，适合研究场景

ROS2 (Robot Operating System 2)

定位: 面向生产环境和实际部署
设计目标: 实时性、可靠性、跨平台
通信机制: 基于DDS (Data Distribution Service) 标准
节点发现: 去中心化，无需Master节点
实时性: 支持实时系统，适合工业应用

核心联系

尽管有重大变化，ROS2仍然：

保持ROS的核心概念: 节点、话题、服务、参数
兼容ROS的消息格式: .msg、.srv、.action文件格式相同
延续ROS的生态系统: 工具链和开发流程相似
提供迁移路径: 官方提供迁移指南和工具

关键差异对比表

特性	ROS1	ROS2
通信中间件	TCPROS (自定义)	DDS (标准)
节点发现	集中式 (ROS Master)	去中心化 (DDS Discovery)
实时性	不支持	支持 (QoS配置)
跨平台	主要Linux	Linux/Windows/macOS
Python版本	Python 2.7	Python 3.x
C++标准	C++03/11	C++14/17
构建系统	catkin	ament
Launch格式	XML	Python
参数系统	全局参数服务器	节点级参数
时间API	rospy.Time	rclpy.time
日志系统	rospy.log*	node.get_logger()

架构层面的根本变化

1. 通信架构：从TCPROS到DDS

ROS1的通信机制

节点A ──TCPROS──> ROS Master <──TCPROS── 节点B
         (注册)                    (查询)
         ↓                          ↓
    建立直接连接 <─────────────────────┘

特点:

需要ROS Master作为中介
Master故障会导致整个系统崩溃
通信延迟不可控
不支持QoS配置

ROS2的通信机制

节点A ──DDS──┐
             ├──> DDS Domain ──> 节点B
节点C ──DDS──┘

特点:

无需Master节点，去中心化
基于DDS标准，支持多种实现（Fast DDS, RTI Connext等）
支持QoS配置（可靠性、持久性、截止时间等）
更好的实时性能

QoS示例:

# ROS2中可以为每个订阅者/发布者配置QoS
from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy

qos_profile = QoSProfile(
    reliability=ReliabilityPolicy.RELIABLE,  # 可靠传输
    durability=DurabilityPolicy.TRANSIENT_LOCAL,  # 持久化
    depth=10  # 队列深度
)

publisher = node.create_publisher(
    String, 'topic', qos_profile
)

2. 节点发现机制

ROS1：集中式发现

# ROS1需要先启动roscore
# 所有节点向Master注册
rospy.init_node('my_node')  # 自动连接到Master

问题:

Master是单点故障
网络分区时无法工作
启动顺序依赖

ROS2：去中心化发现

# ROS2无需Master，节点自动发现
rclpy.init()
node = rclpy.create_node('my_node')  # 自动发现其他节点

优势:

无单点故障
支持网络分区
启动顺序无关

3. 命名空间系统

ROS1

# 全局命名空间
rospy.Publisher('/global/topic', ...)

# 相对命名空间
rospy.Publisher('relative/topic', ...)

# 私有命名空间
rospy.Publisher('~private/topic', ...)  # 展开为 /node_name/private/topic

ROS2

# 命名空间通过节点创建时指定
node = Node('my_node', namespace='robot1')

# 话题自动添加命名空间
publisher = node.create_publisher(String, 'topic')  
# 实际话题: /robot1/topic

# 无前导斜杠的全局话题
publisher = node.create_publisher(String, '/global/topic')

关键差异: ROS2中命名空间管理更严格，需要显式指定。

4. 参数系统

ROS1：全局参数服务器

# 所有节点共享一个全局参数服务器
rospy.set_param('/global_param', value)
value = rospy.get_param('/global_param')
value = rospy.get_param('~private_param', default)  # 节点私有参数

特点:

全局可见
动态修改
无类型检查

ROS2：节点级参数

# 每个节点有自己的参数
class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        # 必须先声明参数
        self.declare_parameter('param_name', default_value)
        self.declare_parameter('int_param', 10)
        self.declare_parameter('string_param', 'default')
        
        # 然后才能获取
        value = self.get_parameter('param_name').get_parameter_value().double_value
        int_val = self.get_parameter('int_param').get_parameter_value().integer_value

特点:

节点隔离
类型安全
必须先声明后使用
支持参数描述和约束

5. 时间系统

ROS1

import rospy

now = rospy.Time.now()
duration = rospy.Duration(1.0)  # 1秒
future = now + duration

# 时间对象可以直接运算
if time1 < time2:
    pass

ROS2

from rclpy.time import Time, Duration

now = self.get_clock().now()
duration = Duration(seconds=1.0, nanoseconds=0)
future = now + duration

# 时间比较
if time1 < time2:
    pass

差异:

ROS2时间对象更类型安全
需要从clock获取时间
Duration构造更明确

代码层面的迁移改动

Python代码迁移 (rospy → rclpy)

1. 节点初始化

ROS1:

#!/usr/bin/env python
import rospy

rospy.init_node('my_node', anonymous=True)
rospy.spin()

ROS2:

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')

def main():
    rclpy.init()
    node = MyNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

关键变化:

必须使用类继承Node
需要显式调用rclpy.init()和rclpy.shutdown()
anonymous参数不再需要（节点名唯一性由DDS保证）

2. 发布者和订阅者

ROS1:

pub = rospy.Publisher('topic', MessageType, queue_size=10)
sub = rospy.Subscriber('topic', MessageType, callback, queue_size=10)

ROS2:

# 在Node类中
self.pub = self.create_publisher(MessageType, 'topic', 10)
self.sub = self.create_subscription(MessageType, 'topic', callback, 10)

关键变化:

方法名：Publisher → create_publisher
参数顺序：(topic, type, queue) → (type, topic, queue)
必须在Node类的方法中调用

3. 服务

ROS1 - 服务端:

def callback(request):
    response = ServiceResponse()
    response.result = request.data * 2
    return response

service = rospy.Service('service_name', ServiceType, callback)

ROS1 - 客户端:

client = rospy.ServiceProxy('service_name', ServiceType)
response = client(request_data)  # 同步阻塞调用

ROS2 - 服务端:

def callback(request, response):
    response.result = request.data * 2
    return response

service = self.create_service(ServiceType, 'service_name', callback)

ROS2 - 客户端:

self.client = self.create_client(ServiceType, 'service_name')

# 等待服务可用
if not self.client.wait_for_service(timeout_sec=1.0):
    self.get_logger().error('Service not available')
    return

# 异步调用
request = ServiceType.Request()
request.data = 10
future = self.client.call_async(request)

# 等待响应
rclpy.spin_until_future_complete(self, future)
if future.done():
    response = future.result()

关键变化:

服务回调签名变化：(request) → (request, response)
客户端调用变为异步：call() → call_async()
必须使用spin_until_future_complete等待响应

4. 参数处理

ROS1:

# 直接获取，无需声明
value = rospy.get_param('~param_name', default_value)
rospy.set_param('param_name', new_value)

# 获取参数列表
param_names = rospy.get_param_names()

ROS2:

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        # 必须先声明参数
        self.declare_parameter('param_name', default_value)
        self.declare_parameter('int_param', 10)
        self.declare_parameter('string_param', 'default')
        
    def use_params(self):
        # 获取参数（需要指定类型）
        value = self.get_parameter('param_name').get_parameter_value().double_value
        int_val = self.get_parameter('int_param').get_parameter_value().integer_value
        str_val = self.get_parameter('string_param').get_parameter_value().string_value
        
        # 设置参数
        from rclpy.parameter import Parameter
        self.set_parameters([
            Parameter('param_name', Parameter.Type.DOUBLE, new_value)
        ])
        
        # 获取所有参数
        params = self.get_parameters(['param_name', 'int_param'])

关键变化:

必须先声明后使用（在__init__中）
参数有类型（double, int, string, bool等）
获取参数需要指定类型
设置参数需要创建Parameter对象

5. 时间处理

ROS1:

import rospy

now = rospy.Time.now()
duration = rospy.Duration(1.0)  # 1秒
future = now + duration

# 创建定时器
rospy.Timer(rospy.Duration(1.0), callback)  # 1Hz

# 速率控制
rate = rospy.Rate(10)  # 10 Hz
rate.sleep()

ROS2:

from rclpy.duration import Duration

# 获取当前时间
now = self.get_clock().now()

# 创建时长
duration = Duration(seconds=1.0, nanoseconds=0)
future = now + duration

# 创建定时器
self.create_timer(1.0, callback)  # 1秒，自动转换为Duration

# 速率控制
rate = self.create_rate(10)  # 10 Hz
rate.sleep()

关键变化:

时间从clock对象获取，不是全局函数
Duration构造更明确（seconds和nanoseconds）
Timer和Rate通过Node方法创建

6. 日志系统

ROS1:

rospy.loginfo('Info message')
rospy.logwarn('Warning message')
rospy.logerr('Error message')
rospy.logdebug('Debug message')
rospy.logfatal('Fatal message')

ROS2:

self.get_logger().info('Info message')
self.get_logger().warn('Warning message')
self.get_logger().error('Error message')
self.get_logger().debug('Debug message')
self.get_logger().fatal('Fatal message')

关键变化:

通过logger对象调用，不是全局函数
方法名小写（loginfo → info）

7. TF变换

ROS1:

import tf

listener = tf.TransformListener()
try:
    (trans, rot) = listener.lookupTransform(
        '/target_frame',
        '/source_frame',
        rospy.Time(0)  # 最新变换
    )
except (tf.LookupException, tf.ConnectivityException, tf.ExtrapolationException):
    pass

ROS2:

from tf2_ros import TransformListener, Buffer
import tf2_geometry_msgs

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        self.tf_buffer = Buffer()
        self.tf_listener = TransformListener(self.tf_buffer, self)
    
    def get_transform(self):
        try:
            transform = self.tf_buffer.lookup_transform(
                'target_frame',
                'source_frame',
                rclpy.time.Time()  # 最新变换
            )
            return transform
        except Exception as e:
            self.get_logger().error(f'Transform error: {e}')
            return None

关键变化:

tf → tf2_ros
需要Buffer和TransformListener对象
异常处理简化

C++代码迁移 (roscpp → rclcpp)

1. 节点初始化

ROS1:

#include <ros/ros.h>

int main(int argc, char** argv) {
    ros::init(argc, argv, "my_node");
    ros::NodeHandle nh;
    ros::NodeHandle nh_private("~");
    
    ros::spin();
    return 0;
}

ROS2:

#include <rclcpp/rclcpp.hpp>

int main(int argc, char** argv) {
    rclcpp::init(argc, argv);
    auto node = std::make_shared<rclcpp::Node>("my_node");
    rclcpp::spin(node);
    rclcpp::shutdown();
    return 0;
}

关键变化:

头文件：ros/ros.h → rclcpp/rclcpp.hpp
使用智能指针管理节点
显式调用shutdown()

2. 发布者和订阅者

ROS1:

ros::Publisher pub = nh.advertise<std_msgs::String>("topic", 10);
ros::Subscriber sub = nh.subscribe("topic", 10, callback);

ROS2:

auto pub = node->create_publisher<std_msgs::msg::String>("topic", 10);
auto sub = node->create_subscription<std_msgs::msg::String>(
    "topic", 10, callback);

关键变化:

消息类型命名空间：std_msgs::String → std_msgs::msg::String
方法名：advertise → create_publisher

3. 服务

ROS1:

bool callback(ServiceType::Request& req, ServiceType::Response& res) {
    res.result = req.data * 2;
    return true;
}

ros::ServiceServer service = nh.advertiseService("service_name", callback);

ROS2:

void callback(
    const std::shared_ptr<ServiceType::Request> request,
    std::shared_ptr<ServiceType::Response> response) {
    response->result = request->data * 2;
}

auto service = node->create_service<ServiceType>("service_name", callback);

关键变化:

回调使用智能指针
返回类型：bool → void
方法名：advertiseService → create_service

构建系统的变化

package.xml迁移

ROS1:

<?xml version="1.0"?>
<package format="2">
  <name>my_package</name>
  <version>1.0.0</version>
  
  <buildtool_depend>catkin</buildtool_depend>
  
  <depend>roscpp</depend>
  <depend>rospy</depend>
  <depend>std_msgs</depend>
  
  <build_depend>message_generation</build_depend>
  <exec_depend>message_runtime</exec_depend>
</package>

ROS2:

<?xml version="1.0"?>
<package format="3">
  <name>my_package</name>
  <version>1.0.0</version>
  
  <buildtool_depend>ament_cmake</buildtool_depend>
  
  <depend>rclcpp</depend>
  <depend>rclpy</depend>
  <depend>std_msgs</depend>
  
  <build_depend>rosidl_default_generators</build_depend>
  <exec_depend>rosidl_default_runtime</exec_depend>
  <member_of_group>rosidl_interface_packages</member_of_group>
</package>

关键变化:

format="2" → format="3"
catkin → ament_cmake
roscpp → rclcpp
rospy → rclpy
message_generation → rosidl_default_generators
message_runtime → rosidl_default_runtime
消息包需要<member_of_group>rosidl_interface_packages</member_of_group>

CMakeLists.txt迁移

ROS1:

cmake_minimum_required(VERSION 2.8.3)
project(my_package)

find_package(catkin REQUIRED COMPONENTS
  roscpp
  rospy
  std_msgs
  message_generation
)

add_message_files(
  FILES
  MyMessage.msg
)

generate_messages(
  DEPENDENCIES
  std_msgs
)

catkin_package(
  CATKIN_DEPENDS roscpp rospy std_msgs message_runtime
)

ROS2:

cmake_minimum_required(VERSION 3.5)
project(my_package)

find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)
find_package(rclpy REQUIRED)
find_package(std_msgs REQUIRED)
find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/MyMessage.msg"
  DEPENDENCIES std_msgs
)

ament_export_dependencies(
  rclcpp
  rclpy
  std_msgs
)

ament_package()

关键变化:

CMake版本：2.8.3 → 3.5
find_package(catkin) → find_package(ament_cmake)
add_message_files() + generate_messages() → rosidl_generate_interfaces()
catkin_package() → ament_export_dependencies() + ament_package()

Launch文件迁移

ROS1 (XML格式):

<launch>
  <arg name="robot_name" default="robot1"/>
  
  <group ns="$(arg robot_name)">
    <node name="node1" pkg="my_package" type="node1.py"/>
    <node name="node2" pkg="my_package" type="node2" output="screen"/>
    
    <rosparam file="$(find my_package)/config/params.yaml"/>
    <param name="param1" value="value1"/>
  </group>
  
  <include file="$(find other_package)/launch/other.launch"/>
</launch>

ROS2 (Python格式):

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, GroupAction, IncludeLaunchDescription
from launch_ros.actions import Node
from launch_ros.parameter_descriptions import ParameterValue
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.substitutions import FindPackageShare
import os

def generate_launch_description():
    robot_name_arg = DeclareLaunchArgument(
        'robot_name',
        default_value='robot1'
    )
    
    node1 = Node(
        package='my_package',
        executable='node1.py',
        name='node1',
        namespace=LaunchConfiguration('robot_name')
    )
    
    node2 = Node(
        package='my_package',
        executable='node2',
        name='node2',
        namespace=LaunchConfiguration('robot_name'),
        output='screen'
    )
    
    # 加载参数文件
    params_file = PathJoinSubstitution([
        FindPackageShare('my_package'),
        'config',
        'params.yaml'
    ])
    
    return LaunchDescription([
        robot_name_arg,
        GroupAction([
            node1,
            node2,
        ]),
        # 包含其他launch文件
        IncludeLaunchDescription(
            PathJoinSubstitution([
                FindPackageShare('other_package'),
                'launch',
                'other.launch.py'
            ])
        )
    ])

关键变化:

格式: XML → Python（ROS2不支持XML launch文件）
<launch> → generate_launch_description()函数
<arg> → DeclareLaunchArgument
<node> → Node
<group> → GroupAction
<rosparam> → 参数文件通过Node的parameters参数加载
<include> → IncludeLaunchDescription

迁移策略与注意事项

迁移顺序

消息包 - 最基础，无依赖
基础功能包 - 不依赖ROS的包
ROS功能包 - 按依赖关系顺序
应用包 - 依赖其他包的包
元包 - 最后迁移

关键注意事项

1. 参数必须先声明

❌ 错误:

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        # 直接使用参数会失败
        value = self.get_parameter('param').value

✅ 正确:

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        # 必须先声明
        self.declare_parameter('param', default_value)
        # 然后才能使用
        value = self.get_parameter('param').value

2. 服务调用是异步的

❌ 错误:

future = client.call_async(request)
response = future.result()  # 可能为None，因为还没完成

✅ 正确:

future = client.call_async(request)
rclpy.spin_until_future_complete(self, future)
if future.done():
    response = future.result()

3. Launch文件必须转换

⚠️ ROS2不支持XML格式的launch文件，所有.launch文件必须转换为.launch.py。

4. Python版本要求

ROS1: Python 2.7
ROS2: Python 3.x

所有Python脚本需要：

更新shebang: #!/usr/bin/env python → #!/usr/bin/env python3
检查Python 2/3兼容性问题
print语句 → print()函数

5. 命名空间处理

ROS2中命名空间更严格：

节点创建时指定命名空间
话题自动添加命名空间前缀
全局话题需要前导斜杠

6. 时间API变化

时间从clock对象获取
Duration构造更明确
时间运算方式相同，但类型不同

7. 消息类型命名空间

ROS1:

from std_msgs.msg import String

ROS2:

from std_msgs.msg import String  # 相同
# 但服务消息：
from package.srv import ServiceType

8. 编译系统

ROS1: catkin_make 或 catkin build
ROS2: colcon build

工作空间结构相同，但构建命令不同。

迁移检查清单

每个包迁移后检查：

[ ] package.xml 已更新为format="3"
[ ] CMakeLists.txt 已更新为ament_cmake
[ ] 所有Python代码已迁移（rospy → rclpy）
[ ] 所有C++代码已迁移（roscpp → rclcpp）
[ ] 所有launch文件已转换为Python格式
[ ] 参数系统正确使用（先声明后使用）
[ ] 服务调用正确处理异步
[ ] 时间API正确使用
[ ] TF迁移正确（tf → tf2）
[ ] 日志系统正确使用
[ ] 编译通过
[ ] 基本功能测试通过

总结

ROS1 vs ROS2 核心差异

维度	ROS1	ROS2
设计目标	研究原型	生产部署
通信机制	TCPROS	DDS
节点发现	集中式(Master)	去中心化
实时性	不支持	支持(QoS)
跨平台	主要Linux	全平台
构建系统	catkin	ament
Launch格式	XML	Python
参数系统	全局	节点级
Python版本	2.7	3.x