ROS2 高级接口与中间件
10 TF2 坐标变换
10 TF2 坐标变换(TF2 Transform)
10.1 TF2 概述
10.1.1 什么是 TF2
TF2(Transform 2)是 ROS 2 中用于管理坐标变换的库。它跟踪多个坐标系之间的关系,并允许开发人员在不同坐标系之间转换数据(例如点、向量、姿态)。
10.1.2 TF2 的使用
| 应用 | 注释 |
|---|---|
| 传感器集成 | 将不同传感器的数据转换到统一的坐标系 |
| 导航 | 将地图坐标系转换为机器人坐标系。 |
| 机械臂 | 计算终端执行器相对于基座的姿态 |
| 可视化 | 在 RViz2 中正确显示机器人状态 |
10.1.3 坐标系命名
| 名称 | 用途 |
|---|---|
| Map | 全局/世界坐标系,固定不变 |
| odom | 用于定位的里程计坐标系。 |
| base_link | 机器人基座坐标系。 |
| base_footprint | 机器人底盘投影到地面。 |
| Camera_link | 摄像头坐标系 |
| Laser_link | 激光雷达坐标系 |
23. ROS2 TF2 坐标变换
TF2 简介
坐标系是一个非常熟悉的概念,也是机器人学的重要基础。在一个完整的机器人系统中会有许多坐标系,这些坐标的位置应如何管理?ROS 为我们提供了一个坐标管理器:TF2
TF 系统参考:tf: The transport library
2. 机器人中的坐标系
坐标系在移动机器人系统中同样重要,例如,移动机器人的中心点是 base_link,雷达的位置称为雷达坐标系即 laser_link,机器人的移动里程,以累积的方式计算,称为 odom,这反过来又有累积误差和漂移,绝对位置称为地图坐标 Map。
一层坐标之间的关系很复杂,有些是相对固定的,有些是不断变化的,看似简单的坐标在空间中变得复杂,一个良好的坐标系系统尤其重要。

关于坐标系变换的基本理论,在每本机器人学教材中都有解释,可分解为平移和旋转部分,通过 4×4 矩阵描述,在空间中绘制坐标系,并在两者之间交替,这实际上是向量的数学描述。
ROS 中 TF 函数的底层原理是封装这些数学变换,详细的理论知识可在所有机器人学课程中获得,我们主要解释如何使用 TF 坐标管理系统。
3 TF 命令行操作
让我们从两只小海龟的例子开始,了解一下基于坐标的机器人计算。为了便于演示,本节课程更适合在虚拟机上操作。
3.1. 工具的安装
此示例要求我们首先安装功能包、tf 海龟仿真器、tf 树可视化工具。
sudo apt install ros-${ROS_DISTRO}-turtle-tf2-py ros-humble-tf2-tools
sudo pip3 install transforms3d
sudo apt install ros-${ROS_DISTRO}-rqt-tf-tree3.2. 启动
然后我们可以从 launch 文件开始,然后我们可以控制其中一只小海龟,另一只将自动跟随移动。打开两个终端运行以下命令:
# ros2 launch turtle_tf2_py turtle_tf2_demo.launch.py
ros2 run turtlesim turtle_teleop_key当我们控制一只海龟移动时,另一只海龟会跟随它。
3.3. 查看 TF 树
ros2 run rqt_tf_tree rqt_tf_tree可以在 rqt 窗口中看到 TF 变换树
3.4. 查询坐标变换信息
如果我们想知道一个或两个坐标之间的具体关系,我们可以通过这个工具 tf2_echo 看到:
ros2 run tf2_ros tf2_echo turtle2 turtle1一旦运行成功,终端将循环显示坐标系的变换。
3.5. 坐标的可视化
运行 rviz2,然后添加 TF 显示插件
rviz2rivz2 中的参考坐标是:world,添加 TF 显示,如果海龟移动,Rviz 中的轴将开始移动,这样是不是更直观?
4. 静态坐标变换
所谓静态坐标变换是指两个坐标之间的相对位置是固定的。例如,雷达和 base_link 之间的位置是固定的。
示例:为了演示,本节课程更适合在虚拟机上操作
4.1. A 到 B 位置的发布
ros2 run tf2_ros static_transform_publisher 0 0 3 0 0 3.14 A B4.2. 拦截/访问 TF 关系
ros2 run tf2_ros tf2_echo A B4.3. rivz 可视化
运行 rviz2,然后添加 TF 显示插件
rviz25. 案例展示
上节课中说明了系统提供给海龟的 TF 关系,我们自己来做。
课程内容:
小海龟跟随案例的编程
掌握动态广播者的编程
控制坐标转换之间监听坐标的编程
通过 PID 控制将物理量(范围、角度)受控转换为速度控制
进度:
理解 TF 系统的跨时间维度变换坐标功能
海龟跟随案例的实现原理分析
在两个海龟仿真器中,我们可以定义三个坐标系,如仿真器的全局参考系称为 world,海龟 1 和海龟 2 的坐标,在两只海龟的中心,这样海龟 1 与世界坐标的相对位置代表海龟 1 的位置,海龟 2 也是同样。
为了让海龟二朝海龟一移动,我们将在两者之间建立连接,添加箭头。怎么样?我们说坐标是通过向量转换的,所以在这个跟随例程中,这是一个用 TF 解决的好方法。
向量的长度表示距离,方向表示角度,距离和角度,这样我们就可以通过设置一个时间来计算速度,然后是速度话题的发布和发布,海龟可以移动。
所以这个例程的核心是通过坐标系计算向量,两只海龟将持续移动,这个向量必须在给定的周期上计算,这将需要使用 TF 的实时广播和监听。
7. 新建功能包
在 ~/workspace/src 目录中创建一个新的功能来存储我们的文件
ros2 pkg create pkg_tf --build-type ament_python --dependencies rclpy --node-name turtle_tf_broadcaster完成上述命令后,将创建 pkg_tf 包,并将创建一个用于 turtle_tf_broadcaster 的节点,相关配置文件已配置,将以下代码添加到 turtle_tf_broadcaster.py 文件中:
import math
import rclpy # ROS 2 Python client library
from rclpy.node import Node # ROS 2 node class
from geometry_msgs.msg import TransformStamped # transform message
from tf2_ros import TransformBroadcaster # TF transform broadcaster
from turtlesim.msg import Pose # turtlesim pose message
def quaternion_from_euler(roll, pitch, yaw):
"""Return quaternion from Euler angles (roll, pitch, yaw)."""
cy = math.cos(yaw * 0.5)
sy = math.sin(yaw * 0.5)
cp = math.cos(pitch * 0.5)
sp = math.sin(pitch * 0.5)
cr = math.cos(roll * 0.5)
sr = math.sin(roll * 0.5)
w = cr * cp * cy + sr * sp * sy
x = sr * cp * cy - cr * sp * sy
y = cr * sp * cy + sr * cp * sy
z = cr * cp * sy - sr * sp * cy
return (x, y, z, w)
class TurtleTFBroadcaster(Node):
def __init__(self, name):
super().__init__(name)
self.turtlename = self.declare_parameter('turtlename', 'turtle').value
self.tf_broadcaster = TransformBroadcaster(self)
self.subscription = self.create_subscription(
Pose,
f'/{self.turtlename}/pose',
self.turtle_pose_callback, 1)
def turtle_pose_callback(self, msg):
transform = TransformStamped()
transform.header.stamp = self.get_clock().now().to_msg()
transform.header.frame_id = 'world'
transform.child_frame_id = self.turtlename
transform.transform.translation.x = msg.x
transform.transform.translation.y = msg.y
transform.transform.translation.z = 0.0
q = quaternion_from_euler(0, 0, msg.theta)
transform.transform.rotation.x = q[0]
transform.transform.rotation.y = q[1]
transform.transform.rotation.z = q[2]
transform.transform.rotation.w = q[3]
self.tf_broadcaster.sendTransform(transform)
def main(args=None):
rclpy.init(args=args)
node = TurtleTFBroadcaster("turtle_tf_broadcaster")
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()- 在下一个 turtle_tf_broadcaster.py 目录下添加以下代码:
import math
import rclpy
from rclpy.node import Node
import tf_transformations
from tf2_ros import TransformException
from tf2_ros.buffer import Buffer
from tf2_ros.transform_listener import TransformListener
from geometry_msgs.msg import Twist
from turtlesim.srv import Spawn
class TurtleFollowing(Node):
def init(self, name):
super().init(name)
self.declare_parameter('source_frame', 'turtle1')
self.source_frame = self.get_parameter(
'source_frame').get_parameter_value().string_value
self.tf_buffer = Buffer()
self.tf_listener = TransformListener(self.tf_buffer, self)
self.spawner = self.create_client(Spawn, 'spawn')
self.turtle_spawning_service_ready = False
self.turtle_spawned = False
self.publisher = self.create_publisher(Twist, 'turtle2/cmd_vel', 1)
self.timer = self.create_timer(1.0, self.on_timer)
def on_timer(self):
from_frame_rel = self.source_frame
to_frame_rel = 'turtle2'
if self.turtle_spawning_service_ready:
if self.turtle_spawned:
try:
now = rclpy.time.Time()
trans = self.tf_buffer.lookup_transform(
to_frame_rel,
from_frame_rel,
now)
except TransformException as ex:
self.get_logger().info(
f'Could not transform {to_frame_rel} to {from_frame_rel}: {ex}')
return
msg = Twist()
scale_rotation_rate = 1.0
msg.angular.z = scale_rotation_rate * math.atan2(
trans.transform.translation.y,
trans.transform.translation.x)
scale_forward_speed = 0.5
msg.linear.x = scale_forward_speed * math.sqrt(
trans.transform.translation.x ** 2 +
trans.transform.translation.y ** 2)
self.publisher.publish(msg)
else:
if self.result.done():
self.get_logger().info(
f'Successfully spawned {self.result.result().name}')
self.turtle_spawned = True
else:
self.get_logger().info('Spawn is not finished')
else:
if self.spawner.service_is_ready():
request = Spawn.Request()
request.name = 'turtle2'
request.x = float(4)
request.y = float(2)
request.theta = float(0)
self.result = self.spawner.call_async(request)
self.turtle_spawning_service_ready = True
else:
self.get_logger().info('Service is not ready')
def main(args=None):
rclpy.init(args=args)
node = TurtleFollowing("turtle_following")
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()在 pkg_tf 包下创建新的 launch 文件夹,并在 launch 文件夹中创建新的 turtle_following.launch.py 文件,添加以下内容:
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import Node
def generate_launch_description():
return LaunchDescription([
DeclareLaunchArgument('source_frame', default_value='turtle1', description='Target frame name.'),
Node(
package='turtlesim',
executable='turtlesim_node',
),
Node(
package='pkg_tf',
executable='turtle_tf_broadcaster',
name='broadcaster1',
parameters=[
{'turtlename': 'turtle1'}
]
),
Node(
package='pkg_tf',
executable='turtle_tf_broadcaster',
name='broadcaster2',
parameters=[
{'turtlename': 'turtle2'}
]
),
Node(
package='pkg_tf',
executable='turtle_following',
name='listener',
parameters=[
{'source_frame': LaunchConfiguration('source_frame')}
]
),
])8. 编辑配置文件
8.1. setup.py 配置
导入相关库
import os
from glob import glob添加 turtle_following 节点信息并添加将 launch 文件复制到 install 共享目录的命令
(os.path.join('share',package_name,'launch'),glob('launch/*')),9. 编译功能包
colcon build --packages-select pkg_tf操作步骤
刷新终端环境变量,然后运行
# source install/setup.bash
ros2 launch pkg_tf turtle_following.launch.py激活海龟键盘控制节点,控制第一只小海龟移动,第二只将自动跟随。
ros2 run turtlesim turtle_teleop_key在该终端中按上下键可控制其中一只小海龟的运动,另一只将跟随,直到它们重叠。
11. 进度
理解 TF 的跨时间变换
Buffer 能够通过缓冲区自动缓存过去 10s 内 TF 系统中的所有 TF 转换(可以通过 Buffer 自由设置任何持续时间),缓冲区中的所有变化都按时间打上时间戳,所有变化都可以连续追踪,即使两个坐标在不同的时间点,也可以追溯彼此。这可以参考设计 TF 系统的文献,有详细的原理(文献链接在本节课程文件夹下或本节开头)。
下面是我们的海龟跟随我们的情况的补充,其中红色箭头表示可以随着时间的推移在不同的时间点搜索两个坐标之间的变化。
自定义接口消息
11 自定义接口消息
在 ROS 系统中,Topic、Service 和 Action 三种通信机制都依赖于一个核心概念:通信接口。
通信本质是多方信息交换,而非单向自我表达。为了实现高效可靠的交互,参与通信的节点必须对数据的格式和语义有共同的理解。为此,ROS 引入了标准化的通信接口,为各类消息定义了清晰统一的数据结构,确保不同节点之间传输的信息得到准确理解。
此接口设计不仅规范了数据交换方式,还在结构层面装饰了程序模块:开发者无需了解彼此的内部实现,只需遵循接口协议即可实现模块之间的无缝协作。这允许集成他人开发的功能组件并重用自己的代码,从而显著提高开发效率。
最后,通信接口是 ROS"避免重复造轮子"概念的技术构建块——通过标准化和解耦促进机器人软件的模块化、重用和生态系统开发。
ROS 有三种常见的通信机制:topics、services、actions,通过每个定义的接口,各种节点有机地连接。
11.1 创建自定义接口流程
主要步骤如下:
创建接口功能包
创建并编辑 .msg 文件、.srv 文件、.action 文件
编辑配置文件
编译
测试
11.2 为动作通信创建自定义接口
在 09 动作通信案例中,我们已经演示过,如果您创建动作通信接口的完整流程,您可以回顾一下,我们在这里不再重复。
11.3 为话题通信创建自定义接口
我们已在 09 动作新闻通信中创建了自定义接口套件,现在我们正在功能包 pkg_interfaces 下创建一个新的 msg 文件夹,并在 msg 文件夹下创建一个新的 Person.msg 文件,输入以下内容:
string name
int32 age
float64 height将以下配置添加到 package.xml 和 CMakeLists.txt:
# CMakeLists.txt
rosidl_generate_interfaces(${PROJECT_NAME}
"action/Progress.action"
"msg/Person.msg"
)package.xml
# package.xml
<buildtool_depend>rosidl_default_generators</buildtool_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<depend>action_msgs</depend>
<member_of_group>rosidl_interface_packages</member_of_group>终端进入当前工作空间,编译功能包:
cd ~/workspace
colcon build --packages-select pkg_interfaces测试接口正常
首先刷新环境变量
# source install/setup.bash查看接口类型
ros2 interface show pkg_interfaces/msg/Person正常情况下,终端将导出与 Person.msg 文件一致的内容。
11.4 为服务通信创建自定义接口
在 [ROS2 动作通信服务实现] 课程中,我们已经创建了自定义接口功能包,在 pkg_interfaces 包下新建一个 srv 文件夹,在 srv 文件夹下新建一个 Add.srv 文件,在该文件中输入以下内容:
int32 num1
int32 num2
---
int32 sum将以下配置添加到 package.xml 和 CMakeLists.txt:
CMakeLists.txt
rosidl_generate_interfaces(${PROJECT_NAME}
"action/Progress.action"
"msg/Person.msg"
"srv/Add.srv"
)package.xml
<buildtool_depend>rosidl_default_generators</buildtool_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<depend>action_msgs</depend>
<member_of_group>rosidl_interface_packages</member_of_group>终端进入当前工作空间,编译功能包:
cd ~/workspace
colcon build --packages-select pkg_interfaces
source install/setup.bash测试
ros2 interface show pkg_interfaces/srv/Add正常情况下,终端将导出与 Person.msg 文件一致的内容。
11.5 后续步骤
1.12 参数服务案例 - 学习参数服务
2.13 元功能包 - 学习包
12 参数服务案例
12 参数服务(Parameters)
12.1 参数概述
在 ROS 机器人系统中,参数充当 C++ 全局变量,并为多个节点提供简单的数据共享机制。这些参数以全局字典的形式存储在系统中——即所谓的字典,即由"键"(参数名)和"值"(参数数据)组成的映射关系,类似于编程语言中的变量赋值(参数名 = 参数值),并且只能通过名称访问。
参数系统具有强分布式特性:一旦参数被声明或更新,其他节点不仅可以实时访问数据,还可以通过监控机制确保整个系统保持最新。这种设计实现了节点间的无缝数据协作,并保持全局一致性,无需复杂的点对点通信。
12.2 小海龟例程中的参数
仿真器在小海龟例程中也提供了许多参数,通过这些参数熟悉参数的含义和命令行的使用方式。
在 Jetson 上启动两个终端运行海龟仿真和键盘控制节点:
ros2 run turtlesim turtlesim_node
# Second terminal
ros2 run turtlesim turtle_teleop_key启动另一个终端,通过以下命令查看参数列表
ros2 param list参数查询和修改
如果您想查询或修改参数的值,可以在 Param 命令后跟随 Get 或 Set 子命令:
ros2 param describe turtlesim background_b # view the description of a parameter
ros2 param get turtlesim background_b # query the value of a parameter
ros2 param set turtlesim background_b 10 # modify the value of a parameter参数文件保存和加载
参数的查询/修改太麻烦,尝试参数文件。ROS 参数文件采用 Yaml 格式,可以在 param 命令后跟随 dump 子命令,将节点的所有参数保存到文件中,或通过 load 命令一次性加载参数文件的所有内容:
ros2 param dump turtlesim >> turtlesim.yaml # save a node parameter set to a parameter file
ros2 param load turtlesim turtlesim.yaml # load all parameters from a file at once12.3 参数案例
12.3.1 新建功能包
在 workspace 中的 src 目录下新建功能包
ros2 pkg create pkg_param --build-type ament_python --dependencies rclpy --node-name param_demo12.3.2 代码实现
编辑器编辑 Param_demo.py 通过添加以下代码执行发布者功能:
import rclpy
from rclpy.node import Node
class ParameterNode(Node):
def __init__(self, name):
super().__init__(name)
self.timer = self.create_timer(2.0, self.timer_callback)
self.declare_parameter('robot_name', 'muto')
def timer_callback(self):
robot_name_param = self.get_parameter('robot_name').get_parameter_value().string_value
self.get_logger().info('Hello %s!' % robot_name_param)
def main(args=None):
rclpy.init(args=args)
node = ParameterNode("param_declare")
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()12.3.3 编译功能包
colcon build --packages-select pkg_param12.3.4 操作步骤
首先刷新环境变量,然后运行节点
# source install/setup.bash
ros2 run pkg_param param_demo打开另一个终端,将 robot_name 设置为 robot:
ros2 param set param_declare robot_name robot可以在终端中找到日志信息,muto 是我们默认设置的参数值。参数名称是 robot_name。一旦您通过命令行更改此参数,终端就会更改。
12.4 后续步骤
1.13 元功能包 - 学习元功能包
2.14 分布式通信 - 学习分布式通信
13 元功能包
13 元功能包(Metapackages)
13.1 元功能包摘要
13.1.1 什么是元功能包
在 ROS2 中,一个完整的功能模块通常由多个功能包组成。在机器人导航的情况下,该模块通常包含多个子功能包,例如地图服务、定位算法、路径规划、运动控制等。如果用户需要手动安装这些分散的包,不仅效率低,而且容易因依赖项缺失而导致系统功能失效。
为了解决这个问题,ROS2 引入了元功能包(Metapackage)机制。该概念源自 Linux 的文档管理系统,本质上是一个"假包"——它本身不包含任何实质性代码或节点,而是通过声明依赖将一组相关功能包有机地集成在一起。它可以被理解为功能集群的"目录索引":它清楚地指示该模块包含哪些子包,并指导包管理工具自动完成批量安装。
典型的应用场景是 ROS2 的安装顺序:
sudo apt install ros-humble-desktop这里的 ros-humble-desktop 是一个依赖于数十个包的元功能包,如 ROS2 核心工具、常用库和仿真组件,执行此命令将允许整个系统一次性部署。
在机器人领域,Navigation 2 是元功能包的经典实践。通过包结构,该仓库将十几个独立的导航组件,如 AMCL 定位、代价地图、规划器、控制器等,集成到统一的模块中。完整的导航能力仓库将通过安装 nav2_bringup 包由开发人员自动获取,大大简化了复杂系统的部署过程。
元功能包不直接提供软件,而是依赖其他相关包来为完整包提供简单的安装机制。
13.1.2 元功能包的角色
为了用户友好的安装,我们只需要这个包来组织其他相关包。
| 用途 | 注释 |
|---|---|
| 组织 | 分组相关功能包 |
| 简化安装 | 一次安装多个包 |
| 依赖管理 | 统一管理依赖项 |
| 文档 | 清晰的项目结构 |
| 版本控制 | 统一版本的发布和管理 |
13.2 实现案例
新建功能包
ros2 pkg create pkg_metapackage修改 package.xml 文件以添加执行依赖的包
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
<name>pkg_metapackage</name>
<version>0.0.0</version>
<description>TODO: Package description</description>
<maintainer email="1461190907@qq.com">root</maintainer>
<license>TODO: License declaration</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<exec_depend>pkg_interfaces</exec_depend>
<exec_depend>pkg_helloworld_py</exec_depend>
<exec_depend>pkg_topic</exec_depend>
<exec_depend>pkg_service</exec_depend>
<exec_depend>pkg_action</exec_depend>
<exec_depend>pkg_param</exec_depend>
<test_depend>ament_lint_auto</test_depend>
<test_depend>ament_lint_common</test_depend>
<export>
<build_type>ament_cmake</build_type>
</export>
</package>CMakeLists.txt 文档读取如下:
cmake_minimum_required(VERSION 3.5)
project(pkg_metapackage)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
find_package(ament_cmake REQUIRED)
ament_package()编译功能包
将不会有实际的可实现文档。
colcon build --packages-select pkg_metapackage13.3 后续步骤
您可以:
1.14 分布式通信 - 学习分布式通信
DDS - 学习 DDS 中间件
回顾系列:
| 章节 | 内容 |
|---|---|
| 04 工作空间 | 工作空间管理 |
| 05 包 | 功能包基础 |
| 12 参数服务案例 | 参数配置 |
14 分布式通信
14 分布式通信
14.1 分布式通信摘要
ROS2 是一个强大的分布式通信框架,允许便捷的主机间网络数据交互。底层基于 DDS(Data Distribution Service)中间件,它通过 ROS_DOMAIN_ID 管理通信:当不同设备上的节点设置相同的域 ID 并处于同一网络时,它们会自动检测并自由通信;反之,ID 之间相互分离。为了简化操作,ROS2 默认所有节点的域 ID 为 0,这意味着您不需要任何额外的配置,只要设备处于同一网络,您就可以获得即开即用的分布式消息。此功能在需要多设备数据交互的场景中具有广泛和关键的应用,如无人机编队、无人机群和远程控制。
14.1.1 什么是分布式通信
ROS 2 分布式通信允许多台计算机节点在没有中央服务器的情况下相互通信。这是通过 DDS(Data Distribution Service)实现的。
14.1.2 分布式通信的特性
| 特性 | 注释 |
|---|---|
| 无中心化 | 不需要 ROS Master。 |
| 自动发现 | 节点自动找到网络上的其他节点 |
| 跨平台 | 不同操作系统之间的通信 |
| 可靠传输 | 支持多种 QoS 策略 |
14.2 ROS_DOMAIN_ID
14.2.1 域 ID 概念
ROS_DOMAIN_ID 用于隔离不同的 ROS2 网络。具有相同域 ID 的节点可以相互通信,具有不同域 ID 的节点彼此隔离。
14.3 实现
14.3.1 默认实现
通过简单地将主机和操作员 [多个] 放在同一网络中实现分布式通信。例如,主机连接到相同的 WiFi 或相同的路由器。
Windows 中的虚拟网络与主机处于同一网络中。
测试:
假设我们有两台主机 A 和 B,它们可以以任何形式连接到网络,例如虚拟机、树莓派、jetson、x86,只需相同版本的 ros2 环境。
- 主机 A 执行:
这里的演示是汽车在 docker 中,docker 模型是主机模型,它只是与汽车共享网络,所以它与汽车实现没有区别。
ros2 run demo_nodes_py talker- 主机 B 执行:
ros2 run demo_nodes_py listener如下所示:主机端话题立即被订阅,表明已实现多机通信
14.3.2 分布式网络分组
假设您与其他机器人在同一网络中,您可以为您的机器人设置组以避免其他机器人的干扰。
ROS2 提供了 DOMAIN 机制,与子组类似,能够与同一 DOMAIN 中的计算机通信,我们可以在主机 [汽车] 和从机 [虚拟机] 中添加一句。
$ export ROS_DOMAIN_ID=<your_domain_id>如果主机 [车] 与从机 [虚拟机] 分配的 ID 不同,则两者无法以分组方式通信。
14.3.3 案例 1
- 主机 [车] 执行:
echo "export ROS_DOMAIN_ID=6" >> ~/.bashrc # Here `6` is the `ROS_DOMAIN_ID`; it does not have to be `6` as long as it follows the `ROS_DOMAIN_ID` rules
source ~/.bashrc
ros2 run demo_nodes_py talker2 同时,从 [虚拟机]:
echo "export ROS_DOMAIN_ID=6" >> ~/.bashrc # Keep this value the same as the host side
source ~/.bashrc
ros2 run demo_nodes_py listener案例 2
通过分布式通信控制小海龟移动
主机 A 运行命令
ros2 run turtlesim turtlesim_node主机 B 运行命令
ros2 run turtlesim turtle_teleop_key14.4 注意事项
设置 ROS_DOMAIN_ID 的值时,它不是随机的,但也是有约束的:
推荐 ROS_DOMAIN_ID 值在 [0,101] 之间,包含 0 和 101。
每个域 ID 内的节点总数有限制,需要小于或等于 120;
如果域 ID 为 101,则该域中节点的总数需要小于或等于 54。
14.5 DDS 域 ID 值的计算规则(分级知识)
域 ID 值的相关计算规则如下:
如果 DDS 基于 TCP/IP 或 UDP/IP 网络通信协议,则为网络通信指定端口号,端口号以两字节整数、无符号整数表示,值范围在 [0,65535] 之间;
端口号的分配也受其规则约束,不能任意使用,DDS 协议下以 7400 为起始端口,即可用端口为 [7400,65535],已知,在 DDS 协议下默认每个域 ID 占用 250 个端口,域 ID 数量为:(65535-7400)/250 = 232,对应的值范围为 [0,231];
操作系统也会设置一些预占端口,DDS 中使用时也需要避免这些以避免使用冲突和不同操作系统的预占差异,最终结果是在 Linux 下可用域 ID 为 [0,101] 和 [215-231],而 Windows 和 Mac 中可用域 ID 为 [0,166]。综上所述,为了多平台兼容,推荐域 ID 在 [0,101] 范围内取值。
每个域 ID 默认占用 250 个端口,每个 ROS2 节点需要两个端口。此外,第 1 个和第 2 个端口是 Discovery Multicast 和 User Multicast 端口,从第 11 个和第 12 个端口开始,Discovery Unicast 和 User Unicast 端口,以及后续节点占用的端口依次扩展,域 ID 中节点的最大数量为:(250-10)/2 = 120(一个);
特殊情况:当域 ID 值为 101 时,后续端口的一半是操作系统的预占端口,最多 54 个节点。
以上计算规则就足够了。
14.6 后续步骤
DDS - 深入学习 DDS 中间件
时间相关 API - 学习 Time API
https://fast-dds.docs.eprosima.com/en/latest/
15 DDS
15 DDS(Data Distribution Service)
15.1 DDS 概述
15.1.1 什么是 DDS?
DDS(Data Distribution Service)是一种以数据为中心的发布-订阅中间件标准,ROS 2 使用 DDS 实现底层通信。
15.1.2 DDS 核心功能
| 功能 | 注释 |
|---|---|
| 发现机制 | 自动找到网络上的 DDS 参与者 |
| 发布/订阅 | 解决数据传输模式 |
| QoS 策略 | 可用的服务质量 |
| 类型系统 | 强类型数据定义 |
| 零拷贝 | 高效数据传输 |
15.2 通信模型
我们在前面课程中学习的话题、服务、动作,以及它们的底层通信的实际实现,都是由 DDS 完成的,这相当于 ROS 机器人系统的神经网络。
DDS 的核心是通信,有大量可以实现通信的模型和软件框架,这里我们列出四种常用的模型。
首先,点对点模型:多个客户端直接连接到同一服务。每次通信,双方都需要建立链接;随着节点数量的增加,连接数也增加。同时,每个客户端必须清楚地知道服务端的确切地址和它提供的服务。一旦服务端地址改变,所有客户端将被迫修改,影响很大。
第二,Broker 模型:它是基于点对点模型的改进。所有请求都被统一分配给 Broker,Broker 负责转发并识别真正能提供服务的节点。因此,客户端不再需要关心服务器地址。然而,它的问题也很突出:Broker 处于系统中心,处理能力对整体效率有直接影响,在系统规模扩大时容易成为性能瓶颈。更糟的是,如果 Broker 失败,整个系统可能崩溃。ROS1 采用了类似的结构。
第三,广播模型:所有节点可以在同一通道上发送广播消息,所有节点都可以接收。这种方法避免了依赖服务器地址的问题,通信双方之间不需要单独连接。但缺点也很明显:路由上的信息量非常大,所有节点都必须处理每条消息,其中绝大多数实际上与自己无关。
第四,以数据为中心的 DDS 模型:这种方法在某种程度上类似于广播模型,节点可以在 DataBus 上发布或订阅数据。但它更先进,因为通信中有多个并行的数据访问路由,每个节点只需关注自己感兴趣的数据,可以直接忽略。这就像一个旋转的盘子,所有的菜都在 DataBus 上传递,我们只需要拿我们想要的,其余的完全被忽略。
可以看出,在这些通信模型中,DDS 的优点更加明显。
15.3 DDS 在 ROS2 中的应用
DDS 在 ROS2 中的位置至关重要,所有上层都建立在 DDS 之上。在这个 ROS2 中,蓝色和红色部分是 DDS。
在 ROS 的四个组件中,通过加入 DDS,分布式通信系统的集成显著增强,因此我们在开发机器人时不必处理通信,可以将更多时间用于在其他部分开发应用程序。
15.4 服务质量策略
DDS 的基础设施是 Domain。Domain 用于组织应用程序以完成通信。回想一下,我们在树莓派和计算机互通时配置的 DOMAIN_ID,本质上是全局数据空间的分组:只在同一 DOMAIN 组的节点上,我们才能发现并相互通信。这样,可以有效避免不连接的数据。
DDS 的另一个核心特性是服务质量策略:Qos。
QoS 可以理解为基于网络的传输规则:应用程序将声明其期望的传输质量行为,而 QoS 机制负责尽可能满足这些要求。这就像数据发布者和订阅者之间的"通信合同"。
策略如下:
DEADLINE 策略:表示每个数据通信必须在规定的截止时间内至少完成一次;
HISTORY 策略:表示历史数据缓存的大小限制;
RELIABILITY 策略:表示数据传输的可靠模式。如果配置为 BEST_EFFORT(尽可能),即使网络状况不佳,数据流也尽可能良好,但数据可能丢失;如果配置为 RELIABLE(可靠传输),通信期间数据完整性尽可能得到确保,例如图像传输,不太可能缺失。我们可以根据实际应用场景选择合适的模型。
DURABILITY 策略:可以配置为为后期到达的节点提供历史数据,允许新节点更快地进入系统。
15.5. 测试案例
案例 1 - 通过命令行配置 DDS
打开第一个终端,使用以下命令发布话题:
ros2 topic pub /chatter std_msgs/msg/Int32 "data: 66" --qos-reliability best_effort再次打开终端使用不同的 Qos 打印话题,如果 Qos 策略与发布者不同,将出现警告,主题数据将无法正常接收:
ros2 topic echo /chatter --qos-reliability reliable我们使用与话题发布者相同的 Qos 策略来接收主题数据。